Build MCP

🚀 MCP開發保姆級教程：從零搭建到上線部署，一條龍搞定！

MCP作為一項新興技術，在當下十分火爆。中文互聯網上關於如何開發MCP服務的資料眾多，但大多語焉不詳或淺嘗輒止，很多案例只是照搬官方文檔示例寫的水文。

作為開發者，我深知工程化對於MCP服務開發的重要性。其開發並非簡單地編寫幾個服務接口，還需全面考慮代碼結構、配置管理、日誌記錄、異常處理等方面。

經過不斷探索和實踐，我將MCP服務的開發流程整理成了這份詳盡的指南，希望能助力更多開發者快速上手，構建出高質量的MCP服務。

本項目代碼已開源至 GitHub ，歡迎大家 Star 和 Fork。

🚀 快速開始

為了讓大家有更明確的實踐目標，我們將簡單構建一個基於高德地圖API的MCP服務，該服務具備以下核心功能：

根據用戶ip獲取用戶的地理位置

• 參數：ip地址(可選，不傳遞默認獲取當前主機IP)
• 返回：用戶的經緯度信息

根據用戶的地理位置獲取附近的POI信息

• 參數：經度、緯度、POI類型
• 返回：附近的POI信息列表

完成上述功能後，我們就可以通過大模型對話獲取真實的POI信息。例如：

• 用戶：我想知道我附近的餐館有哪些？
• MCP服務：根據您的位置，附近有以下餐館：1. 餐館A 2. 餐館B 3. 餐館C
• 用戶：餐館A的地址是什麼？
• MCP服務：餐館A的地址是：
  • 地址：XXX
  • 電話：XXX
  • 營業時間：XXX

效果圖

✨ 主要特性

MCP服務器可以提供三種主要類型的功能：

• Resources（資源）：客戶端可以讀取的類似文件的數據（如 API 響應或文件內容）
• Tools（工具）：LLM 可以調用的函數（經用戶批准）
• Prompts（提示）：幫助用戶完成特定任務的預先編寫的模板

📦 安裝指南

必備知識

在開始之前，建議您具備以下知識：

• Python 基礎
• LLM（大語言模型）概念
• UV （Python 包管理工具）

系統要求

• Python 3.10 及以上版本
• Python MCP SDK 1.2.0

配置開發環境

⚠️ 重要提示

請務必根據自己的操作系統調整命令，powershell 和 bash 的命令語法有所不同。

作者使用的是windows+git終端。本教程前半段與官方基本無異，可查考官方文檔中server開發示例。

安裝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

💻 使用示例

基礎用法

本項目代碼示例豐富，涵蓋了配置管理、日誌模塊、高德地圖請求SDK、MCP服務主程序等多個方面。以下是部分基礎代碼示例：

配置管理代碼

# 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

日誌模塊代碼

# 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:
  """
  獲取一個帶文件和控制檯輸出的 logger。

  Args:
      name (str): logger 名稱，默認為 "default"。
      max_bytes (int): 單個日誌文件最大大小，默認為 5MB。
      backup_count (int): 日誌文件保留份數，默認為 3。
  Returns:
      logging.Logger: 配置好的 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代碼

# 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): 日誌記錄器，默認使用模塊 logger。
  """

  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

📚 詳細文檔

規劃項目目錄結構（推薦 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%（來源)

編寫工具代碼

規劃好目錄後我們開始正式進行編碼，一個正式規範的項目可能涉及到非常多的項目配置讀取。首先第一步我們對配置文件讀取功能進行封裝。

使用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:
  """
  獲取一個帶文件和控制檯輸出的 logger。

  Args:
      name (str): logger 名稱，默認為 "default"。
      max_bytes (int): 單個日誌文件最大大小，默認為 5MB。
      backup_count (int): 日誌文件保留份數，默認為 3。
  Returns:
      logging.Logger: 配置好的 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文檔，我們需要實現兩個主要功能：

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): 日誌記錄器，默認使用模塊 logger。
  """

  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=你的key uv run pytest -s tests/services/test_gd_sdk.py

🚀 MCP 服務的三種傳輸協議簡介

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)

📊 協議對比概覽

協議	通信方向	使用場景	特性亮點	推薦程度
stdio	雙向（本地）	本地子進程調用	簡單、低延遲、零網絡依賴	⭐ 本地優選
SSE	單向（服務器→客戶端）	早期遠程實現	實現簡單，但不支持恢復	⚠️ 已棄用
streamable‑http	雙向/可選 SSE 推送	雲端/遠程交互	單端點、多功能、斷點續傳、兼容性強	✅ 推薦使用

編寫 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"Locating IP: {ip}")
  try:
    result = await sdk.locate_ip(ip)
    if not result:
      ApiResponse.fail("定位結果為空，請檢查日誌，系統異常請檢查相關日誌，日誌默認路徑為/var/log/build_mcp。")
    logger.info(f"Locate IP result: {result}")
    return ApiResponse.ok(data=result, meta={"ip": ip})
  except Exception as e:
    logger.error(f"Error locating 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"Searching nearby: 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"Search nearby result: {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"Error searching nearby: {e}")
    return ApiResponse.fail(str(e))

代碼中我們封裝了統一的響應類，提供了兩個工具函數：

• locate_ip：根據 IP 地址獲取地理位置
• search_nearby：根據經緯度和關鍵詞進行周邊搜索

⚠️ 重要提示

代碼中Annotated類型是必不可少的，這樣能讓LLM通過元信息更加精準地調用工具。目前看到大部分開發者開發的MCP服務都沒有這種意識，只是單純地定義工具，其實效果非常糟糕。

💡 使用建議

同時我們編寫了一個prompt，這個prompt會提供在對話上下文中，是非常重要的一點，也是很多開發者並沒有意識到的。AI時代，我們不僅要寫得好代碼，更要學會如何對提示詞進行打磨。

其實文章主要核心在以上這部分代碼，請認真去理解這部分信息。

至此，我們已經完成了 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():
    """Main function to run the MCP server."""
    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"🚀 Starting MCP server with transport type: %s", args.transport)

    try:
        mcp.run(transport=args.transport)
    except (KeyboardInterrupt, asyncio.CancelledError):
        logger.info("🛑 MCP Server received shutdown signal. Cleaning up...")
    except Exception as e:
        logger.exception("❌ MCP Server crashed with unhandled exception: %s", e)
    else:
        logger.info("✅ MCP Server shut down cleanly.")


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=你的KEY mcp dev src/build_mcp/__init__.py

編寫Makefile

為了方便開發和測試，我們可以編寫一個Makefile來管理常用的命令。

touch Makefile

# Makefile for MCP Service

# 默認目標 - 顯示幫助信息
.DEFAULT_GOAL := help

# 項目環境變量
API_KEY ?= your_api_key_here  # 默認測試用的 API_KEY

# 安裝項目依賴
install:
	@echo "Installing project dependencies..."
	uv pip install -e .

# 運行測試 (需要設置 API_KEY)
test:
	@echo "Running tests with API_KEY=$(API_KEY)..."
	API_KEY=$(API_KEY) uv run pytest -s tests

# 啟動 stdio 協議的 MCP 服務
stdio:
	@echo "Starting MCP service with stdio protocol..."
	uv run build_mcp

# 啟動 streamable-http 協議的 MCP 服務
http:
	@echo "Starting MCP service with streamable-http protocol..."
	uv run build_mcp streamable-http

# dev
dev:
	@echo "Starting MCP service with stdio protocol in development mode..."
	API_KEY=$(API_KEY) mcp dev src/build_mcp/__init__.py

# 別名目標
streamable-http: http

# 幫助信息
help:
	@echo "MCP Service Management"
	@echo ""
	@echo "Usage:"
	@echo "  make install        Install project dependencies"
	@echo "  make test           Run tests (set API_KEY in Makefile or override)"
	@echo "  make stdio          Start MCP service with stdio protocol"
	@echo "  make http           Start MCP service with streamable-http protocol"
	@echo ""
	@echo "Advanced:"
	@echo "  Override API_KEY:   make test API_KEY=custom_key"
	@echo "  Clean:              make clean"
	@echo "  Full setup:         make setup"

# 清理項目
clean:
	@echo "Cleaning project..."
	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 "Project setup completed!"

# 聲明偽目標
.PHONY: install test stdio http streamable-http help clean setup

目前為止我們已經從0到1完整開發完了一整個MCP服務，恭喜自己又學會了一個新技能！

如何使用這個MCP服務？

首先你得擁有一個MCP客戶端，目前市場上各種類型得MCP客戶端層出不窮，至於用什麼全憑你的愛好了。

這裡有一份非常詳細的MCP客戶端使用攻略，是github上一個非常棒的項目：MCP客戶端使用攻略

選擇一個客戶端下載安裝，然後我們對我們開發的服務進行配置。

配置Stdio協議的MCP服務

{
    "mcpServers": {
        "build_mcp": {
            "command": "uv",
            "args": [
                "run",
                "-m"
                "build_mcp"
            ],
            "env": {
                "API_KEY": "你的高德API Key"
            }
        }
    }
}

⚠️ 重要提示

要注意本地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客戶端連接到我們的服務

行文至此結束，祝大家學習愉快！如果你有任何問題或建議，請提交issue或pull request，到GitHub倉庫

參考資料

• Model Context Protocol (MCP) 官方文檔
• 高德地圖API文檔
• 官方開發SDK