Build MCP

🚀 MCP開発の初心者向けチュートリアル：ゼロからの構築とデプロイまで

MCPの人気は言うまでもありません。この新興技術に関する開発資料は、中文インターネット上に数多く存在しますが、多くは説明が不十分であったり、表面的な内容に留まっていたりします。多くのケースは公式ドキュメントのサンプルをそのまま引用しただけの記事です。

開発者として、私はエンジニアリングの重要性を深く理解しています。MCPサービスの開発は、単にいくつかのサービスインターフェースを書くだけではなく、コード構造、設定管理、ログ記録、例外処理など、様々な側面を考慮する必要があります。

このプロジェクトでは、MCPサービスの開発プロセスを詳細なガイドにまとめ、より多くの開発者が迅速にMCPサービスの開発を始め、高品質なサービスを構築できるように支援したいと思います。

本プロジェクトのコードはGitHubで公開されています。皆さんのStarとForkをお待ちしています。

🚀 クイックスタート

構築するMCPサービスの概要

このチュートリアルでは、高德地図APIを使用したMCPサービスを構築します。このサービスは以下の核心機能を備えています。

• ユーザーのIPアドレスから地理位置を取得する

  • パラメータ：IPアドレス（オプション、指定しない場合は現在のホストのIPが使用されます）
  • 戻り値：ユーザーの緯度と経度情報

• ユーザーの地理位置から近くのPOI情報を取得する

  • パラメータ：経度、緯度、POIタイプ
  • 戻り値：近くのPOI情報のリスト

これらの機能を実装することで、大規模言語モデルとの対話を通じて、実際のPOI情報を取得することができます。例えば：

• ユーザー：私の近くにあるレストランを教えてください。
• MCPサービス：あなたの位置に基づくと、近くには以下のレストランがあります。1. レストランA 2. レストランB 3. レストランC
• ユーザー：レストランAの住所は何ですか？
• MCPサービス：レストランAの住所は以下の通りです。
  • 住所：XXX
  • 電話番号：XXX
  • 営業時間：XXX

実行結果の画像

核心MCP概念

MCPについて全く知識がない場合は、MCP公式ドキュメントを読んで基本概念を理解することをおすすめします。

MCPサーバーは以下の3種類の主要な機能を提供することができます。

• Resources：リソース。クライアントが読み取ることができるファイルのようなデータ（API応答やファイル内容など）
• Tools：ツール。LLMが（ユーザーの承認を得て）呼び出すことができる関数
• Prompts：プロンプト。特定のタスクを完了するための事前に作成されたテンプレート

必要な知識

このチュートリアルを始める前に、以下の知識があることをおすすめします。

• Pythonの基礎知識
• LLM（大規模言語モデル）の概念
• UV（Pythonパッケージ管理ツール）

システム要件

• Python 3.10以上のバージョン
• Python MCP SDK 1.2.0

開発環境の設定

⚠️ 重要提示

オペレーティングシステムに合わせてコマンドを調整してください。PowershellとBashのコマンド構文は異なります。

このチュートリアルでは、Windows+Gitターミナルを使用しています。前半部分は公式ドキュメントのサーバー開発サンプルとほぼ同じです。

UVのインストール

# Linux or macOS
curl -LsSf https://astral.sh/uv/install.sh | sh

# Windows
powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"

仮想環境の作成とプロジェクトの初期化

# UVを使用してプロジェクトディレクトリを作成し、移動します。
uv init build-mcp
cd build-mcp

# 仮想環境を作成します。
uv venv
source .venv/Scripts/activate

# 関連する依存関係をインストールします。
uv add mcp[cli] httpx pytest

プロジェクトディレクトリ構造の計画（推奨：src/ レイアウト）

build-mcp/
├── src/                         # 核心ソースコードディレクトリ（Pythonパッケージ）
│   └── build_mcp/               # メインパッケージ名前空間
│       ├── __init__.py          # パッケージ初期化ファイル
│       ├── __main__.py          # コマンドラインエントリポイント
│       ├── common/              # 共通機能モジュール
│       │   ├── config.py        # 設定管理
│       │   └── logger.py        # ログシステム
│       └── services/            # ビジネスサービスモジュール
│           ├── gd_sdk.py        # 高德サービス統合
│           └── server.py        # メインサービス実装
├── tests/                       # テストセットディレクトリ
│   ├── common/                  # 共通モジュールのテスト
│   └── services/                # サービスモジュールのテスト
├── docs/                        # プロジェクトドキュメント
│   └── build‑mcp プロジェクト開発ガイド.md # 核心ドキュメント
├── pyproject.toml               # プロジェクトビルド設定
├── Makefile                     # 自動化コマンド管理
└── README.md                    # プロジェクト概要ドキュメント

構造設計の解説

核心設計：src/ レイアウト（主要な利点）

build-mcp/
└── src/
    └── mirakl_mcp/
        ├── ... 

なぜこの構造を採用するのか？

• ✅ インストール環境の分離（核心価値） テスト時にpip installを使用してパッケージをインストールすることを強制し、ソースコードパスを直接参照することを避け、テスト環境とユーザー実行環境を一致させます。
• ✅ 暗黙的なパス依存の防止 開発ディレクトリがsys.pathの先頭にあることによる誤ったインポートを排除します（src/がない従来のレイアウトでよく見られます）。
• ✅ パッケージの安全性 配布ファイルにパッケージ内容が正しく含まれていることを強制的に検証します（欠落したファイルはテストですぐに明らかになります）。
• ✅ 複数環境の一貫性 開発/テスト/本番環境で完全に同じパッケージ構造を使用し、「私の環境では動く」という問題を解消します。

📊 データサポート：PyPA公式調査によると、src/レイアウトを採用したプロジェクトのパッケージングエラー率は63%低下します（出典）

💻 使用例

基本的な使用法

# 以下のコードは、配置ファイルの読み取り機能を pyaml パッケージを使用して実装しています。
# uv add pyyaml コマンドで pyyaml パッケージを追加します。
uv add pyyaml

# 配置管理モジュールを作成します。
mkdir -p src/build_mcp/common
touch src/build_mcp/__init__.py
touch src/build_mcp/common/__init__.py
touch src/build_mcp/common/config.py

# 配置ファイルを作成します。
touch src/build_mcp/config.yaml

src/build_mcp/config.yaml ファイルに以下の内容を追加します。

# 高德地図APIの設定
api_key: test
# 高德地図APIの基本URL
base_url: https://restapi.amap.com
# プロキシ設定
proxy: http://127.0.0.1:10809
# ログレベル
log_level: INFO
# インターフェースのリトライ回数
max_retries: 5
# インターフェースのリトライ間隔時間（秒）
retry_delay: 1
# 指数バックオフ係数
backoff_factor: 2
# ログファイルのパス
log_dir: /var/log/build_mcp

⚠️ 重要提示

config.yaml ファイルは src/build_mcp/ ディレクトリに配置する必要があります。これにより、配置を読み込むときに正しく見つけることができます。この配置ファイルはエンジニアリングの例としてのみ使用されます。本番環境では、APIキーなどの機密情報を直接配置ファイルに記述しないでください。環境変数またはセキュアなストレージサービスを使用することをおすすめします。

高度な使用法

# 配置管理コードを記述します。
# src/build_mcp/common/config.py
import os

import yaml


def load_config(config_file="config.yaml") -> dict:
    """
    配置ファイルを読み込みます。
  
    Args:
        config_file (str): 配置ファイルの名前、デフォルトは "config.yaml"。
    Returns:
        dict: 配置ファイルの内容を返します。
    Example:
        config = load_config("config.yaml")
        print(config)
    """
    # ルートディレクトリを見つけます（config.yaml はルートディレクトリに配置されます）
    base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    config_path = os.path.join(base_dir, config_file)

    with open(config_path, "r", encoding="utf-8") as f:
        config = yaml.safe_load(f)
    return config

コードのインストール

⚠️ 重要提示

初めてコードをインストールするときは、pip install -e . コマンドを使用してください。これにより、現在のディレクトリが編集可能なパッケージとして仮想環境にインストールされます。開発中にコードを変更すると、再インストールすることなくすぐに反映されます。

uv pip install -e .

テストコードの記述

mkdir -p tests/common
touch tests/common/test_config.py

# tests/common/test_config.py
from build_mcp.common.config import load_config


def test_load_config():
    """配置ファイルの読み込み機能をテストします。"""
    config = load_config("config.yaml")
    assert config["api_key"] == "test"
    assert config["log_level"] == "INFO"

テストの実行

uv run pytest tests

ログモジュールの記述

プログラマーにとって、問題を迅速に特定するためには、ログシステムは非常に重要です。優れたプログラマーは、コードを書くだけでなく、ログもうまく書くことができます。ここでは、ログモジュールを簡単にラップして、後で使用しやすくします。

touch src/build_mcp/common/logger.py

ここでは、コンソールとファイルに同時に出力するログシステムを実装し、ログのローテーションとバックアップをサポートします。

# src/build_mcp/common/logger.py
import logging
import os
from logging.handlers import RotatingFileHandler

from build_mcp.common.config import load_config

config = load_config("config.yaml")


def get_logger(name: str = "default", max_bytes=5 * 1024 * 1024, backup_count=3) -> logging.Logger:
  """
  ファイルとコンソールに出力するロガーを取得します。

  Args:
      name (str): ロガーの名前、デフォルトは "default"。
      max_bytes (int): 単一のログファイルの最大サイズ、デフォルトは 5MB。
      backup_count (int): ログファイルの保持数、デフォルトは 3。
  Returns:
      logging.Logger: 設定されたロガーインスタンス。
  Example:
      logger = get_logger("my_logger")
      logger.info("This is an info message.")
  """
  log_level = config.get("log_level", "INFO")
  log_dir = config.get("log_dir", "./logs")
  if isinstance(log_level, str):
    log_level = getattr(logging, log_level.upper(), logging.INFO)

  os.makedirs(log_dir, exist_ok=True)
  log_file = os.path.join(log_dir, f"{name}.log")

  logger = logging.getLogger(name)
  logger.setLevel(log_level)
  logger.propagate = False

  if not logger.hasHandlers():
    console_handler = logging.StreamHandler()
    console_formatter = logging.Formatter('[%(asctime)s] %(levelname)s - %(message)s')
    console_handler.setFormatter(console_formatter)

    file_handler = RotatingFileHandler(log_file, maxBytes=max_bytes, backupCount=backup_count, encoding='utf-8')
    file_formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    file_handler.setFormatter(file_formatter)
    logger.addHandler(file_handler)
    logger.addHandler(console_handler)

  logger.info(f"Logger 初期化完了、ファイルに書き込みます：{log_file}")
  return logger

ここまでで、システムの基本モジュールが構築されました。次に、核心的なサービス機能を実装します。

高德地図リクエストSDKの記述

高德地図APIドキュメントに基づいて、以下の2つの主要な機能を実装する必要があります。

1. ユーザーのIPアドレスから地理位置を取得する
2. 地理位置から近くのPOI情報を取得する

高德地図サービスモジュールの作成

mkdir -p src/build_mcp/services
touch src/build_mcp/services/__init__.py
touch src/build_mcp/services/gd_sdk.py

高德地図サービスコードの記述

# src/build_mcp/services/gd_sdk.py
import asyncio
import logging
from typing import Any

import httpx


class GdSDK:
  """
  GdSDK API 非同期 SDK ラッパー。

  自動リトライと指数バックオフ戦略をサポートします。

  Args:
      config (dict): 設定辞書、例：
          {
              "base_url": "https://restapi.amap.com",
              "api_key": "your_api_key",
              "proxies": {"http": "...", "https": "..."},  # オプション
              "max_retries": 5,
              "retry_delay": 1,
              "backoff_factor": 2,
          }
      logger (logging.Logger, optional): ログ記録器、デフォルトはモジュールのロガーを使用します。
  """

  def __init__(self, config: dict, logger=None):
    self.api_key = config.get("api_key", "")
    self.base_url = config.get("base_url", "").rstrip('/')
    self.proxy = config.get("proxy", None)
    self.logger = logger or logging.getLogger(__name__)
    self.max_retries = config.get("max_retries", 5)
    self.retry_delay = config.get("retry_delay", 1)
    self.backoff_factor = config.get("backoff_factor", 2)

    # 自動的にリクエストヘッダーとプロキシ設定を持つ非同期HTTPクライアントを作成します。
    self._client = httpx.AsyncClient(proxy=self.proxy, timeout=10)

  async def __aenter__(self):
    return self

  async def __aexit__(self, exc_type, exc, tb):
    await self._client.aclose()

  def _should_retry(self, response: httpx.Response = None, exception: Exception = None) -> bool:
    """
    リクエストが失敗した後にリトライするかどうかを判断します。

    Args:
        response (httpx.Response, optional): HTTP レスポンスオブジェクト。
        exception (Exception, optional): リクエスト例外。

    Returns:
        bool: リトライが必要かどうか。
    """
    if exception is not None:
      # ネットワーク例外など、リトライをおすすめします。
      return True

    if response is not None and response.status_code in (429, 500, 502, 503, 504):
      # サーバーエラーまたはリクエスト過多、リトライをおすすめします。
      return True

    # その他の場合はリトライしません。
    return False

  async def _request_with_retry(self, method: str, url: str, params=None, json=None):
    """
    自動リトライと指数バックオフを伴うHTTPリクエストを送信します。

    Args:
        method (str): HTTPメソッド、例： 'GET', 'POST'。
        url (str): リクエストURL。
        params (dict, optional): URLクエリパラメータ。
        json (dict, optional): リクエストボディJSON。

    Returns:
        dict or None: 成功時はJSON解析結果を返し、失敗時は None を返します。
    """
    for attempt in range(self.max_retries + 1):
      try:
        self.logger.info(f"リクエストを送信します：{method} {url}、パラメータ：{params}, JSON：{json}, 試行回数：{attempt + 1}/{self.max_retries + 1}")
        response = await self._client.request(
          method=method,
          url=url,
          params=params,
          json=json,
        )
        self.logger.info(f"レスポンスを受け取りました：{response.status_code} {response.text}")
        if response.status_code in [200, 201]:
          # 成功時にJSONデータを返します。
          return response.json()

        if not self._should_retry(response=response):
          self.logger.error(f"リクエストが失敗し、リトライできません。ステータスコード：{response.status_code}、URL：{url}")
          return None

        self.logger.warning(
          f"リクエストが失敗しました（ステータスコード：{response.status_code}）、"
          f"{attempt + 1}/{self.max_retries} 回目のリトライ、URL：{url}"
        )

      except httpx.RequestError as e:
        self.logger.warning(
          f"リクエスト例外：{str(e)}、"
          f"{attempt + 1}/{self.max_retries} 回目のリトライ、URL：{url}"
        )

      # 最後のリトライでない場合、指数バックオフで待機します。
      if attempt < self.max_retries:
        delay = self.retry_delay * (self.backoff_factor ** attempt)
        await asyncio.sleep(delay)

    self.logger.error(f"すべてのリトライが失敗しました、URL：{url}")
    return None

  async def close(self):
    """
    非同期HTTPクライアントを閉じて、リソースを解放します。
    """
    await self._client.aclose()

  async def locate_ip(self, ip: str = None) -> Any | None:
    """
    IP 位置決めインターフェース
    https://lbs.amap.com/api/webservice/guide/api/ipconfig

    Args:
        ip (str, optional): 照会する IP、空の場合はリクエスト元の公開 IP を使用します。

    Returns:
        dict: 位置決め結果、失敗時は None を返します。
    """
    url = f"{self.base_url}/v3/ip"
    params = {
      "key": self.api_key,
    }
    if ip:
      params["ip"] = ip

    result = await self._request_with_retry(
      method="GET",
      url=url,
      params=params
    )

    if result and result.get("status") == "1":
      return result
    else:
      self.logger.error(f"IP 位置決めに失敗しました: {result}")
      return None

  async def search_nearby(self, location: str, keywords: str = "", types: str = "", radius: int = 1000, page_num: int = 1, page_size: int = 20) -> dict | None:
    """
    周辺検索（新版 POI）
    https://lbs.amap.com/api/webservice/guide/api-advanced/newpoisearch#t4

    Args:
        location (str): 中心点の緯度と経度、形式： "lng,lat"
        keywords (str, optional): 検索キーワード
        types (str, optional): POI 分類
        radius (int, optional): 検索半径（メートル）、最大 50000、デフォルト 1000
        page_num (int, optional): ページ番号、デフォルト 1
        page_size (int, optional): ページあたりの件数、デフォルト 20、最大 25

    Returns:
        dict | None: 検索結果、失敗時は None を返します
    """
    url = f"{self.base_url}/v5/place/around"
    params = {
      "key": self.api_key,
      "location": location,
      "keywords": keywords,
      "types": types,
      "radius": radius,
      "page_num": page_num,
      "page_size": page_size,
    }

    result = await self._request_with_retry(
      method="GET",
      url=url,
      params=params,
    )

    if result and result.get("status") == "1":
      return result
    else:
      self.logger.error(f"周辺検索に失敗しました: {result}")
      return None

このコードでは、以下の機能を実装しています。

• 非同期HTTPリクエスト、自動リトライと指数バックオフをサポート
• locate_ip メソッド：IPアドレスから地理位置を取得する
• search_nearby 周辺検索メソッド：緯度と経度から近くのPOI情報を取得する

高德地図サービステストコードの記述

mkdir -p tests/services
touch tests/services/test_gd_sdk.py

# tests/test_gd_sdk_real.py
import logging
import os

import pytest
import pytest_asyncio

from build_mcp.services.gd_sdk import GdSDK

API_KEY = os.getenv("API_KEY", "your_api_key_here")  # 環境変数から API Key を取得するか、デフォルト値を使用します。


@pytest_asyncio.fixture
async def sdk():
    config = {
        "base_url": "https://restapi.amap.com",
        "api_key": API_KEY,
        "max_retries": 2,
    }
    async with GdSDK(config, logger=logging.getLogger("GdSDK")) as client:
        yield client


@pytest.mark.asyncio
async def test_locate_ip(sdk):
    result = await sdk.locate_ip()
    assert result is not None, "locate_ip が None を返しました"
    assert result.get("status") == "1", f"locate_ip の呼び出しに失敗しました: {result}"
    assert "province" in result, "locate_ip の返り値に province が含まれていません"


@pytest.mark.asyncio
async def test_search_nearby(sdk):
    result = await sdk.search_nearby(
        location="116.481488,39.990464",
        keywords="ガソリンスタンド",
        radius=3000,
        page_num=1,
        page_size=5
    )
    assert result is not None, "search_nearby が None を返しました"
    assert result.get("status") == "1", f"search_nearby の呼び出しに失敗しました: {result}"
    assert "pois" in result, "search_nearby の返り値に pois が含まれていません"

テストの実行

uv run pytest tests/services/test_gd_sdk.py
# 高德API Key がある場合は、以下のコマンドで直接テストを実行できます。
API_KEY=あなたのAPIキー uv run pytest -s tests/services/test_gd_sdk.py

🚀 MCP サービスの3種類の伝送プロトコルの概要

1. stdio

• 通信方式：ローカルプロセス間で標準入力/出力（stdin/stdout）を通じてJSON‑RPCメッセージを双方向に伝送します。
• 適用シーン：ローカルでのツール呼び出しや子プロセス、デスクトップアプリケーションでの軽量な統合などに適しています。
• 利点：遅延が低く、実装が簡単で、ネットワークが不要です。

2. SSE（Server‑Sent Events、サーバー送信イベント）

• 通信方式：HTTPベース。クライアントがPOSTでメッセージを送信し、サーバーがGETでtext/event‑streamを介して単方向にメッセージを送信します。
• 現在の状態：非推奨（deprecated）となっており、MCP v2024‑11‑05からは「streamable‑http」に置き換えられていますが、互換性を維持しています。
• 利点：初期のリモートシーンで、サーバーからの単方向のメッセージ送信が必要な場合に簡単に実装できます。
• 欠点：サーバーからクライアントへの単方向通信のみで、接続の中断復旧に対応していません。

3. streamable‑http

• 通信方式：HTTPベースの双方向伝送。クライアントがPOSTでJSON‑RPCリクエストを送信し、サーバーは一度の応答（JSON）またはストリーミングSSEメッセージを返すことができ、GETでサーバーからのメッセージ送信も可能です。
• サポート機能：
  • 単一の/mcpエンドポイントですべての通信を処理します。
  • セッション管理（Mcp‑Session‑Idを使用）
  • ストリームの中断再開とメッセージの再送信（HTTPの切断復旧でLast‑Event‑IDをサポート）
  • SSEとの後方互換性
• 現在の状態：MCP v2025‑03‑26からデフォルトで推奨されており、クラウドやリモートデプロイメントに適しており、リモートシーンでの選択肢として最適です。(modelcontextprotocol.io)

📊 プロトコルの比較概要

MCPサービスメインプログラムの記述

次に、MCPサービスのメインプログラムを記述し、クライアントのリクエストを処理し、高德地図SDKを呼び出します。

touch src/build_mcp/services/server.py

import os
from typing import Annotated
from typing import Any, Dict, Generic, Optional, TypeVar

from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel
from pydantic import Field

from build_mcp.common.config import load_config
from build_mcp.common.logger import get_logger
from build_mcp.services.gd_sdk import GdSDK

# 環境変数からAPI_KEYを読み取り、ない場合は設定ファイルから読み取ります。
env_api_key = os.getenv("API_KEY")
config = load_config("config.yaml")
if env_api_key:
  config["api_key"] = env_api_key

# FastMCPサービスを初期化します。
mcp = FastMCP("amap-maps", description="高德地図 MCP サービス", version="1.0.0")
sdk = GdSDK(config=config, logger=get_logger(name="gd_sdk"))
logger = get_logger(name="amap-maps")

# 汎用的なAPI応答モデルを定義します。
T = TypeVar("T")


class ApiResponse(BaseModel, Generic[T]):
  success: bool
  data: Optional[T] = None
  error: Optional[str] = None
  meta: Optional[Dict[str, Any]] = None

  @classmethod
  def ok(cls, data: T, meta: Dict[str, Any] = None) -> "ApiResponse[T]":
    return cls(success=True, data=data, meta=meta)

  @classmethod
  def fail(cls, error: str, meta: Dict[str, Any] = None) -> "ApiResponse[None]":
    return cls(success=False, error=error, meta=meta)


# Promptを定義します。
@mcp.prompt(name="assistant", description="高德地図スマートナビゲーションアシスタント、IP位置決め、周辺POI検索などをサポートします。")
def amap_assistant(query: str) -> str:
  return (
    "あなたは高德地図スマートナビゲーションアシスタントで、IP位置決めと周辺POI検索に精通しています。ユーザーの要求に応じてツールを呼び出し、必要な情報を取得してください。\n"
    "## ツールの呼び出し手順：\n"
    "1. `locate_ip` ツールを呼び出して、ユーザーの緯度と経度を取得します。\n"
    "2. 緯度と経度が正常に取得された場合、その緯度と経度を使用して `search_nearby` ツールを呼び出し、検索キーワードを組み合わせて周辺情報を検索します。\n"
    "## 注意事項：\n"
    "- ユーザーに緯度と経度の情報を要求しないでください。`locate_ip` ツールを使用して自動的に取得してください。\n"
    "- ユーザーの要求に緯度と経度の情報が含まれている場合は、その情報を直接使用して周辺検索を行ってください。\n"
    f"ユーザーの要求は以下の通りです。\n\n {query}。\n"
  )


@mcp.tool(name="locate_ip", description="ユーザーのIPアドレスの位置情報を取得し、省市区の緯度と経度などの情報を返します。")
async def locate_ip(ip: Annotated[Optional[str], Field(description="ユーザーのIPアドレス")] = None) -> ApiResponse:
  """
  IPアドレスから位置を特定します。

  Args:
      ip (str): 特定するIPアドレス。

  Returns:
      dict: 位置特定結果を含む辞書。
  """
  logger.info(f"IPを特定しています: {ip}")
  try:
    result = await sdk.locate_ip(ip)
    if not result:
      ApiResponse.fail("位置特定結果が空です。ログを確認してください。システムエラーの場合は、関連するログを確認してください。ログのデフォルトパスは/var/log/build_mcpです。")
    logger.info(f"IP位置特定結果: {result}")
    return ApiResponse.ok(data=result, meta={"ip": ip})
  except Exception as e:
    logger.error(f"IP {ip} の位置特定にエラーが発生しました: {e}")
    return ApiResponse.fail(str(e))


@mcp.tool(name="search_nearby", description="緯度と経度、キーワードを使用して周辺検索を行い、指定された半径内のPOIリストを返します。")
async def search_nearby(
        location: Annotated[str, Field(description="中心点の緯度と経度、形式： 'lng,lat'、例： '116.397128,39.916527'")],
        keywords: Annotated[str, Field(description="検索キーワード、例： 'レストラン'。", min_length=0)] = "",
        types: Annotated[str, Field(description="POI分類コード、複数の分類はカンマで区切ります")] = "",
        radius: Annotated[int, Field(description="検索半径（メートル）、最大50000", ge=0, le=50000)] = 1000,
        page_num: Annotated[int, Field(description="ページ番号、1から始まります", ge=1)] = 1,
        page_size: Annotated[int, Field(description="ページあたりの件数、最大25", ge=1, le=25)] = 20,
) -> ApiResponse:
  """
   周辺検索を行います。

   Args:
       location (str): 中心点の緯度と経度、形式： "lng,lat"。
       keywords (str, optional): 検索キーワード、デフォルトは空。
       types (str, optional): POI分類、デフォルトは空。
       radius (int, optional): 検索半径（メートル）、最大 50000、デフォルトは 1000。
       page_num (int, optional): ページ番号、デフォルトは 1。
       page_size (int, optional): ページあたりの件数、最大 25、デフォルトは 10。

   Returns:
       dict: 検索結果を含む辞書。
  """
  logger.info(f"周辺検索を行っています: location={location}, keywords={keywords}, types={types}, radius={radius}, page_num={page_num}, page_size={page_size}")
  try:
    result = await sdk.search_nearby(location=location, keywords=keywords, types=types, radius=radius, page_num=page_num, page_size=page_size)
    if not result:
      return ApiResponse.fail("検索結果が空です。ログを確認してください。システムエラーの場合は、関連するログを確認してください。ログのデフォルトパスは/var/log/build_mcpです。")
    logger.info(f"周辺検索結果: {result}")
    return ApiResponse.ok(data=result, meta={
      "location": location,
      "keywords": keywords,
      "types": types,
      "radius": radius,
      "page_num": page_num,
      "page_size": page_size
    })
  except Exception as e:
    logger.error(f"周辺検索にエラーが発生しました: {e}")
    return ApiResponse.fail(str(e))

このコードでは、統一された応答クラスをラップし、2つのツール関数を提供しています。

• locate_ip：IPアドレスから地理位置を取得する
• search_nearby：緯度と経度、キーワードを使用して周辺検索を行う

⚠️ 重要提示

コード中のAnnotated型は必須です。これにより、LLMがメタ情報を使用してツールをより正確に呼び出すことができます。現在、多くの開発者が開発したMCPサービスでは、この意識が欠けており、単にツールを定義するだけでは、実際の効果は非常に低いです。

💡 使用アドバイス

同時に、promptを記述しています。このpromptは対話コンテキストに提供され、非常に重要な要素ですが、多くの開発者が意識していない部分です。AI時代には、コードをうまく書くだけでなく、promptの調整方法も学ぶ必要があります。

この部分のコードが本質的な部分ですので、しっかりと理解してください。

ここまでで、MCPサービスの核心機能の実装が完了しました。次に、サービスのエントリポイントを記述し、MCPサービスを起動します。

MCPサービスエントリポイントの記述

touch src/build_mcp/__init__.py

# src/build_mcp/__init__.py
import argparse
import asyncio

from build_mcp.common.logger import get_logger
from build_mcp.services.server import mcp


def main():
    """MCPサーバーを実行するメイン関数。"""
    logger = get_logger('app')

    parser = argparse.ArgumentParser(description="Amap MCP Server")
    parser.add_argument(
        'transport',
        nargs='?',
        default='stdio',
        choices=['stdio', 'sse', 'streamable-http'],
        help='Transport type (stdio, sse, or streamable-http)'
    )
    args = parser.parse_args()

    logger.info(f"🚀 Transport type: %s でMCPサーバーを起動しています", args.transport)

    try:
        mcp.run(transport=args.transport)
    except (KeyboardInterrupt, asyncio.CancelledError):
        logger.info("🛑 MCPサーバーがシャットダウン信号を受け取りました。クリーンアップ中...")
    except Exception as e:
        logger.exception("❌ MCPサーバーが未処理の例外でクラッシュしました: %s", e)
    else:
        logger.info("✅ MCPサーバーが正常にシャットダウンしました。")


if __name__ == "__main__":
    main()

touch src/build_mcp/__main__.py

# src/build_mcp/__main__.py
from build_mcp import main

if __name__ == "__main__":
    main()

pyproject.toml ファイルの修正

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project]
name = "build-mcp"
version = "0.1.0"
description = "MCPサーバーを構築する"
readme = "README.md"
requires-python = ">=3.10"
dependencies = [
  "httpx>=0.28.1",
  "mcp[cli]>=1.9.4",
  "pytest>=8.4.1",
  "pytest-asyncio>=1.0.0",
  "pyyaml>=6.0.2",
]

[tool.pytest.ini_options]
pythonpath = ["src"]
testpaths = ["tests"]

[project.scripts]
build_mcp = "build_mcp.__main__:main"

[tool.hatch.build.targets.wheel]
packages = ["src/build_mcp"]

この中で重要なのは以下の部分です。

[project.scripts]
build_mcp = "build_mcp.__main__:main"

これは、以下の意味を持ちます。

build_mcp コマンドを実行すると、以下のコードが実行されます。

from build_mcp import main
main()

以下のコマンドを使用して、MCPサービスを起動することができます。

• stdioプロトコルのMCPサービスを起動する：

uv run build_mcp

• streamable-httpプロトコルのMCPサービスを起動する：

uv run build_mcp streamable-http

MCPサービスのデバッグ

MCPサービスのデバッグ方法は、サービスの起動方法によって異なります。

1. クライアントコードを記述してstdioプロトコルのMCPサービスをデバッグする

mkdir -p tests/services
touch tests/services/test_mcp_client.py

import pytest
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client


@pytest.mark.asyncio
async def test_mcp_server():
    async with stdio_client(
            StdioServerParameters(command="uv", args=["run", "build_mcp"])
    ) as (read, write):
        print("サーバーを起動しています...")

        async with ClientSession(read, write) as session:
            await session.initialize()
            print("初期化が完了しました")

            tools = await session.list_tools()
            print("利用可能なツール: ", tools)

            assert hasattr(tools, "tools")
            assert isinstance(tools.tools, list)
            assert any(tool.name == "locate_ip" for tool in tools.tools)

テストの実行

API_KEY=あなたのAPI_KEY uv run pytest -s tests/services/test_mcp_client.py 

これはテストコードの例です。必要に応じて、より多くのテストコードを記述して、MCPサービスの機能を検証することができます。

2. Inspectorを使用してテストする

Inspectorは、公式が提供するMCPサービスのデバッグツールです。これを使用すると、ローカルのWebインターフェースを起動し、そのインターフェースからMCPサービスのツールを直接呼び出すことができます。より直感的で使いやすいので、この方法をおすすめします。詳細は公式ドキュメントを参照してください。

# Inspectorを使用してstdioプロトコルのMCPサービスをデバッグする
API_KEY=あなたのAPI_KEY mcp dev src/build_mcp/__init__.py

Makefileの記述

開発とテストを容易にするために、Makefileを記述して、一般的なコマンドを管理することができます。

touch Makefile

# MCPサービスのMakefile

# デフォルトターゲット - ヘルプ情報を表示する
.DEFAULT_GOAL := help

# プロジェクト環境変数
API_KEY ?= your_api_key_here  # デフォルトのテスト用API_KEY

# プロジェクトの依存関係をインストールする
install:
	@echo "プロジェクトの依存関係をインストールしています..."
	uv pip install -e .

# テストを実行する (API_KEYを設定する必要があります)
test:
	@echo "API_KEY=$(API_KEY)でテストを実行しています..."
	API_KEY=$(API_KEY) uv run pytest -s tests

# stdioプロトコルのMCPサービスを起動する
stdio:
	@echo "stdioプロトコルでMCPサービスを起動しています..."
	uv run build_mcp

# streamable-httpプロトコルのMCPサービスを起動する
http:
	@echo "streamable-httpプロトコルでMCPサービスを起動しています..."
	uv run build_mcp streamable-http

# 開発モードでstdioプロトコルのMCPサービスを起動する
dev:
	@echo "開発モードでstdioプロトコルのMCPサービスを起動しています..."
	API_KEY=$(API_KEY) mcp dev src/build_mcp/__init__.py

# エイリアスターゲット
streamable-http: http

# ヘルプ情報
help:
	@echo "MCPサービスの管理"
	@echo ""
	@echo "使用方法:"
	@echo "  make install        プロジェクトの依存関係をインストールする"
	@echo "  make test           テストを実行する (MakefileでAPI_KEYを設定するか、上書きする)"
	@echo "  make stdio          stdioプロトコルでMCPサービスを起動する"
	@echo "  make http           streamable-httpプロトコルでMCPサービスを起動する"
	@echo ""
	@echo "高度な使用方法:"
	@echo "  API_KEYを上書きする:   make test API_KEY=custom_key"
	@echo "  クリーンアップ:         make clean"
	@echo "  完全なセットアップ:     make setup"

# プロジェクトをクリーンアップする
clean:
	@echo "プロジェクトをクリーンアップしています..."
	rm -rf build dist *.egg-info
	find . -name '*.pyc' -exec rm -f {} +
	find . -name '*.pyo' -exec rm -f {} +
	find . -name '__pycache__' -exec rm -rf {} +

# 完全なセットアップ: クリーンアップ + インストール + テスト
setup: clean install test
	@echo "プロジェクトのセットアップが完了しました！"

# 疑似ターゲットを宣言する
.PHONY: install test stdio http streamable-http help clean setup

ここまでで、MCPサービスをゼロから開発することができました。おめでとうございます！新しいスキルを習得しました。

このMCPサービスの使用方法

まず、MCPクライアントが必要です。現在、市場には様々な種類のMCPクライアントがあり、好みに合わせて選択してください。

こちらに非常に詳細なMCPクライアントの使用ガイドがあります。これはGitHub上の素晴らしいプロジェクトです。MCPクライアント使用攻略

クライアントを選択してダウンロード、インストールした後、開発したサービスを設定します。

stdioプロトコルのMCPサービスの設定

{
    "mcpServers": {
        "build_mcp": {
            "command": "uv",
            "args": [
                "run",
                "-m"
                "build_mcp"
            ],
            "env": {
                "API_KEY": "あなたの高德APIキー"
            }
        }
    }
}

⚠️ 重要提示

ローカルのUV環境に注意してください。複数のUVがインストールされている場合、環境が混乱する可能性があります。開発中には、この点に注意する必要があります。

streamable-httpプロトコルのMCPサービスの設定

プロジェクトの起動

make streamable-http

$ make streamable-http
Starting MCP service with streamable-http protocol...
uv run build_mcp streamable-http
[2025-06-26 15:01:33,775] INFO - Logger 初期化完了、ファイルに書き込みます：/var/log/build_mcp\gd_sdk.log
[2025-06-26 15:01:33,839] INFO - Logger 初期化完了、ファイルに書き込みます：/var/log/build_mcp\amap-maps.log
[2025-06-26 15:01:33,847] INFO - Logger 初期化完了、ファイルに書き込みます：/var/log/build_mcp\app.log
[2025-06-26 15:01:33,848] INFO - 🚀 Starting MCP server with transport type: streamable-http
INFO:     Started server process [6064]
INFO:     Waiting for application startup.
[06/26/25 15:01:33] INFO     StreamableHTTP session manager started                                                                                                                                     streamable_http_manager.py:109
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

起動に成功すると、8000ポートでHTTPサービスが起動します。

クライアントの設定

{
    "mcpServers": {
        "build_mcp_http": {
            "url": "http://localhost:8000/mcp"
        }
    }
}

まとめ

この記事では、高德地図のMCPサービスをゼロから構築する方法を紹介しました。以下の内容が含まれています。

• MCPサービスの基本概念と設定
• 高德地図APIを使用してIP位置決めと周辺検索を行う方法
• MCPサービスの核心機能を記述する方法、設定管理、ログシステム、高德地図SDKなど
• MCPサービスのメインプログラムとエントリポイントを記述する方法
• MCPサービスをデバッグする方法、Inspectorの使用、テストコードの記述など
• Makefileを使用してプロジェクトのコマンドを管理する方法
• MCPクライアントを設定してサービスに接続する方法

以上です。皆さんの学習が楽しいものになることを祈っています。何か質問や提案があれば、GitHubリポジトリにissueまたはpull requestを投稿してください。

参考資料

• Model Context Protocol (MCP) 公式ドキュメント
• 高德地図APIドキュメント
• 公式開発SDK