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