Gaode Map POI

Proporciona información de geolocalización y POI (Punto de Interés) cercanos utilizando la API de Gaode Map.

Documentación

Tutorial de desarrollo MCP para principiantes absolutos: ¡De cero al despliegue en producción, todo en uno!

Lo popular que es MCP ya no necesita que lo repita. Como tecnología emergente, hay innumerables recursos en Internet en chino sobre cómo desarrollar servicios MCP, pero la mayoría son vagos o superficiales, y la mayoría de los casos simplemente copian los ejemplos de la documentación oficial en artículos de relleno.

Como desarrollador, conozco bien la importancia de la ingeniería. El desarrollo de servicios MCP no es tan simple como copiar dos interfaces de servicio; también hay que considerar la estructura del código, la gestión de configuración, el registro de logs, el manejo de excepciones y muchos otros aspectos.

Después de explorar y practicar, he organizado el proceso de desarrollo de un servicio MCP en una guía detallada, con la esperanza de ayudar a más desarrolladores a comenzar rápidamente y construir servicios MCP de alta calidad.

El código de este proyecto es de código abierto en GitHub, todos son bienvenidos a darle Star y Fork.

¿Qué tipo de servicio MCP construimos?

Para tener un mejor objetivo de práctica, construiremos un servicio MCP basado en la API de Gaode Map con las siguientes funciones principales:

Obtener la ubicación geográfica del usuario según su IP

Parámetros: ip地址(可选，不传递默认获取当前主机IP)
Retorno: información de latitud y longitud del usuario

Obtener información de POI cercanos según la ubicación geográfica del usuario

Parámetros: 经度、纬度、POI类型
Retorno: lista de información de POI cercanos

Una vez completadas las funciones anteriores, podemos obtener información real de POI a través de conversaciones con el modelo de lenguaje grande. Por ejemplo:

Usuario: Quiero saber qué restaurantes hay cerca de mí.
Servicio MCP: Según su ubicación, hay los siguientes restaurantes cerca: 1. Restaurante A 2. Restaurante B 3. Restaurante C
Usuario: ¿Cuál es la dirección del Restaurante A?
Servicio MCP: La dirección del Restaurante A es:
- Dirección: XXX
- Teléfono: XXX
- Horario de atención: XXX

Captura de pantalla del resultado

Conceptos principales de MCP

Si antes de comenzar no tienes ni idea de qué es MCP, te recomiendo leer primero la documentación oficial de MCP para entender los conceptos básicos.

Un servidor MCP puede proporcionar tres tipos principales de funcionalidades:

Resources: recursos, datos similares a archivos que el cliente puede leer (como respuestas de API o contenido de archivos)
Tools: herramientas, funciones que el LLM puede invocar (con aprobación del usuario)
Prompts: indicaciones, plantillas preescritas que ayudan al usuario a completar tareas específicas

Conocimientos previos necesarios

Antes de comenzar, se recomienda tener los siguientes conocimientos:

Conceptos básicos de Python
Conceptos de LLM (modelo de lenguaje grande)
UV (herramienta de gestión de paquetes de Python)

Requisitos del sistema

Python 3.10 o superior
Python MCP SDK 1.2.0

Configuración del entorno de desarrollo

⚠ Asegúrate de ajustar los comandos según tu sistema operativo; la sintaxis de comandos de powershell y bash es diferente.

El autor utiliza Windows + terminal Git. La primera mitad de este tutorial es básicamente igual a la oficial; puedes consultar el ejemplo de desarrollo de servidor en la documentación oficial.

Instalar 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"

Crear entorno virtual e inicializar el proyecto

# 使用UV创建并进入项目目录
uv init build-mcp
cd build-mcp

# 创建虚拟环境
uv venv
source .venv/Scripts/activate

# 安装相关依赖
uv add mcp[cli] httpx pytest

Planificar la estructura de directorios del proyecto (se recomienda la estructura `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                    # 项目概览文档

Análisis del diseño de la estructura

Diseño principal: estructura `src/` (ventajas clave)

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

¿Por qué adoptar esta estructura?

✅ Entorno de instalación aislado (valor principal)
Al probar, se fuerza la instalación del paquete a través de pip install, evitando referencias directas a la ruta del código fuente, asegurando que el entorno de prueba = entorno de ejecución del usuario
✅ Previene dependencias implícitas de rutas
Elimina importaciones erróneas causadas por el directorio de desarrollo en la primera posición de sys.path (común en estructuras tradicionales sin src/)
✅ Seguridad del empaquetado
Fuerza la verificación de que el contenido del paquete esté correctamente incluido en los archivos de distribución (los archivos faltantes se exponen inmediatamente en las pruebas)
✅ Consistencia entre entornos
Los entornos de desarrollo/pruebas/producción usan exactamente la misma estructura de paquetes, eliminando el problema de "funciona en mi máquina"

📊 Datos de respaldo: Una encuesta oficial de PyPA muestra que los proyectos que adoptan la estructura src/ reducen la tasa de errores de empaquetado en un 63% (fuente)

Escribir el código de las herramientas

Después de planificar los directorios, comenzamos a codificar formalmente. Un proyecto formal y estandarizado puede implicar mucha lectura de configuración del proyecto. El primer paso es encapsular la funcionalidad de lectura de archivos de configuración.

Usar el paquete `pyyaml` para gestionar la configuración

uv add pyyaml

Crear el módulo de gestión de configuración

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

Crear el archivo de configuración

touch src/build_mcp/config.yaml

Agrega el siguiente contenido al archivo 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

⚠ El archivo config.yaml debe colocarse en el directorio src/build_mcp/ para que se pueda encontrar correctamente al cargar la configuración.

Este archivo de configuración es solo un ejemplo de ingeniería; en entornos de producción no escribas información sensible (como claves de API) directamente en el archivo de configuración. Se recomienda usar variables de entorno o servicios de almacenamiento seguro.

Escribir el código de gestión de configuración

# 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

Instalar el código

⚠ Al instalar el código por primera vez, debes usar el comando pip install -e ., que instala el directorio actual como un paquete editable en el entorno virtual. Así, los cambios en el código durante el desarrollo surten efecto inmediatamente sin necesidad de reinstalar.

uv pip install -e .

Escribir código de prueba

Escribir código de prueba lo más detallado posible en el proyecto es un buen hábito. En la ingeniería de proyectos, escribimos pruebas para las funciones principales para garantizar la corrección y estabilidad del código.

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"

Ejecutar las pruebas

uv run pytest tests

Escribir el módulo de logs

Como programador, poder localizar problemas rápidamente es crucial, y el sistema de logs es muy importante. Un buen programador no solo sabe escribir código, sino también escribir logs. Encapsularemos un módulo de logs simple para uso posterior.

touch src/build_mcp/common/logger.py

Aquí implementamos un sistema de logs que genera salida tanto en consola como en archivo, con soporte para rotación y respaldo de logs.

# 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

Hasta ahora, los módulos base del sistema están completos. A continuación, implementaremos las funciones principales del servicio.

Escribir el SDK de solicitudes de Gaode Map

Según la documentación de la API de Gaode Map, necesitamos implementar dos funciones principales:

Obtener la ubicación geográfica según la IP del usuario
Obtener información de POI cercanos según la ubicación geográfica

Crear el módulo de servicio de Gaode Map

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

Escribir el código del servicio de Gaode Map

# 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

El código implementa:

Solicitudes HTTP asíncronas con reintentos automáticos y retroceso exponencial
El método locate_ip para obtener la ubicación geográfica según la IP
El método de búsqueda de alrededores search_nearby para obtener información de POI cercanos según latitud y longitud

Escribir el código de prueba del servicio de Gaode Map

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"

Ejecutar las pruebas

uv run pytest tests/services/test_gd_sdk.py
# 如果你有高德API Key，可以直接运行以下命令进行测试
API_KEY=你的key uv run pytest -s tests/services/test_gd_sdk.py

🚀 Introducción a los tres protocolos de transporte de MCP

1. `stdio`

Método de comunicación: transmisión bidireccional de mensajes JSON‑RPC entre procesos locales a través de entrada/salida estándar (stdin/stdout);
Escenarios de uso: invocación local de herramientas o subprocesos, como integración ligera en aplicaciones de escritorio;
Ventajas: baja latencia, implementación simple, sin necesidad de red.

2. `SSE` (Server‑Sent Events, eventos enviados por el servidor)

Método de comunicación: basado en HTTP: el cliente envía mensajes con POST, y el servidor establece una conexión text/event‑stream unidireccional a través de GET;
Estado actual: está en desuso (deprecated); desde MCP v2024‑11‑05 fue reemplazado por "streamable‑http", pero aún se mantiene compatibilidad;
Ventajas: adecuado para implementaciones simples en escenarios remotos tempranos que solo necesitan push del servidor;
Desventajas: solo unidireccional de servidor a cliente, y la conexión no admite reanudación de puntos de interrupción.

3. `streamable‑http`

Método de comunicación: transmisión bidireccional basada en HTTP: el cliente solicita JSON‑RPC a través de POST, el servidor puede devolver una respuesta única (JSON) o mensajes SSE en streaming, y también puede establecer push del servidor mediante GET;
Funciones compatibles:
- Un único endpoint /mcp maneja toda la comunicación;
- Gestión de sesiones (a través de Mcp‑Session‑Id);
- Reanudación de flujo en puntos de interrupción y reproducción de mensajes (la recuperación de desconexión HTTP admite Last‑Event‑ID);
- Compatibilidad retroactiva con SSE;
Estado actual: recomendado por defecto desde MCP v2025‑03‑26, adecuado para despliegues en la nube y remotos, es la opción preferida para escenarios remotos; (modelcontextprotocol.io)

📊 Resumen comparativo de protocolos

Protocolo	Dirección de comunicación	Escenario de uso	Características destacadas	Nivel de recomendación
stdio	Bidireccional (local)	Invocación de subprocesos locales	Simple, baja latencia, cero dependencia de red	⭐ Preferido local
SSE	Unidireccional (servidor→cliente)	Implementación remota temprana	Implementación simple, pero sin soporte de reanudación	⚠️ En desuso
streamable‑http	Bidireccional/push SSE opcional	Interacción en la nube/remota	Un solo endpoint, multifuncional, reanudación de puntos de interrupción, alta compatibilidad	✅ Recomendado

Escribir el programa principal del servicio MCP

A continuación, escribimos el programa principal del servicio MCP, que maneja las solicitudes del cliente e invoca el SDK de Gaode Map.

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))

En el código encapsulamos una clase de respuesta unificada y proporcionamos dos funciones de herramienta:

locate_ip: obtiene la ubicación geográfica según la dirección IP
search_nearby: realiza búsqueda de alrededores según latitud, longitud y palabras clave

Ten en cuenta que el tipo Annotated en el código es indispensable, ya que permite que el LLM invoque las herramientas con mayor precisión mediante metainformación. Actualmente, la mayoría de los desarrolladores de servicios MCP no tienen esta conciencia; simplemente definen las herramientas, y el resultado es realmente deficiente.

También escribimos un prompt que se proporciona en el contexto de la conversación; esto es muy importante y muchos desarrolladores no lo tienen en cuenta. En la era de la IA, no solo debemos escribir buen código, sino también aprender a pulir los prompts.

En realidad, el núcleo principal del artículo está en esta parte del código; por favor, tómate el tiempo de entender bien esta información.

Con esto, hemos completado la implementación de las funciones principales del servicio MCP. A continuación, necesitamos escribir el punto de entrada del servicio para iniciar el servicio MCP.

Escribir el punto de entrada del servicio 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()

Modificar el archivo `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"]

El punto clave es:

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

Esto significa:

Al ejecutar el comando build_mcp, equivale a ejecutar:

from build_mcp import main
main()

Podemos ejecutar el servicio MCP con los siguientes comandos:

Iniciar el servicio MCP con protocolo stdio:

uv run build_mcp

Iniciar el servicio MCP con protocolo streamable-http:

uv run build_mcp streamable-http

Depurar el servicio MCP

Cómo depurar el servicio MCP depende de la forma en que lo iniciemos.

1. Escribir código de cliente para depurar el servicio MCP con protocolo stdio

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)

Ejecutar las pruebas

API_KEY=你的API_KEY uv run pytest -s tests/services/test_mcp_client.py

Esto es código de prueba; aquí solo es un ejemplo. Puedes escribir más código de prueba según tus necesidades para verificar las funciones del servicio MCP.

2. Usar Inspector para probar

Inspector es una herramienta de depuración de servicios MCP proporcionada oficialmente. Con ella puedes iniciar una interfaz web local donde puedes invocar directamente las herramientas del servicio MCP. Es más intuitiva y fácil de usar; se recomienda este método. Para más detalles, consulta la documentación oficial.

# 使用Inspector调试stdio协议的MCP服务
API_KEY=你的KEY mcp dev src/build_mcp/__init__.py

Escribir el Makefile

Para facilitar el desarrollo y las pruebas, podemos escribir un Makefile para gestionar los comandos comunes.

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

Hasta ahora hemos desarrollado por completo un servicio MCP de principio a fin. ¡Felicidades, has aprendido una nueva habilidad!

¿Cómo usar este servicio MCP?

Primero necesitas tener un cliente MCP. Actualmente hay todo tipo de clientes MCP en el mercado; cuál usar depende completamente de tu preferencia.

Aquí hay una guía muy detallada sobre el uso de clientes MCP, un proyecto excelente en GitHub: Guía de uso de clientes MCP

Elige un cliente, descárgalo e instálalo, y luego configura el servicio que hemos desarrollado.

Configurar el servicio MCP con protocolo Stdio

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

⚠ Presta atención al entorno UV local; si tienes varias versiones de UV instaladas, puede causar confusión en el entorno. Esto es un punto problemático durante el desarrollo, así que ten cuidado.

Configurar el servicio MCP con protocolo Streamable-HTTP

Iniciar el proyecto

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)

Una vez iniciado correctamente, se ejecutará un servicio HTTP en el puerto 8000.

Configuración del cliente

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

Resumen

Este artículo explica cómo construir desde cero un servicio MCP de Gaode Map, cubriendo el siguiente contenido:

Conceptos básicos y configuración del servicio MCP
Cómo usar la API de Gaode Map para geolocalización por IP y búsqueda de alrededores
Cómo escribir las funciones principales del servicio MCP, incluyendo gestión de configuración, sistema de logs y SDK de Gaode Map
Cómo escribir el programa principal y el punto de entrada del servicio MCP
Cómo depurar el servicio MCP, incluyendo el uso de Inspector y la escritura de código de prueba
Cómo usar el Makefile para gestionar los comandos del proyecto
Cómo configurar un cliente MCP para conectarse a nuestro servicio

Con esto concluye el artículo. ¡Que disfruten aprendiendo! Si tienes alguna pregunta o sugerencia, envía un issue o pull request al repositorio de GitHub

Gaode Map POI

Documentación

Tutorial de desarrollo MCP para principiantes absolutos: ¡De cero al despliegue en producción, todo en uno!

El código de este proyecto es de código abierto en GitHub, todos son bienvenidos a darle Star y Fork.

¿Qué tipo de servicio MCP construimos?

Obtener la ubicación geográfica del usuario según su IP

Obtener información de POI cercanos según la ubicación geográfica del usuario

Captura de pantalla del resultado

Conceptos principales de MCP

Conocimientos previos necesarios

Requisitos del sistema

Configuración del entorno de desarrollo

Instalar UV

Crear entorno virtual e inicializar el proyecto

Planificar la estructura de directorios del proyecto (se recomienda la estructura src/)

Análisis del diseño de la estructura

Diseño principal: estructura src/ (ventajas clave)

Escribir el código de las herramientas

Usar el paquete pyyaml para gestionar la configuración

Crear el módulo de gestión de configuración

Crear el archivo de configuración

Escribir el código de gestión de configuración

Instalar el código

Escribir código de prueba

Ejecutar las pruebas

Escribir el módulo de logs

Escribir el SDK de solicitudes de Gaode Map

Crear el módulo de servicio de Gaode Map

Escribir el código del servicio de Gaode Map

Escribir el código de prueba del servicio de Gaode Map

Ejecutar las pruebas

🚀 Introducción a los tres protocolos de transporte de MCP

1. stdio

2. SSE (Server‑Sent Events, eventos enviados por el servidor)

3. streamable‑http

📊 Resumen comparativo de protocolos

Escribir el programa principal del servicio MCP

Escribir el punto de entrada del servicio MCP

Modificar el archivo pyproject.toml

Depurar el servicio MCP

1. Escribir código de cliente para depurar el servicio MCP con protocolo stdio

Ejecutar las pruebas

2. Usar Inspector para probar

Escribir el Makefile

¿Cómo usar este servicio MCP?

Configurar el servicio MCP con protocolo Stdio

Configurar el servicio MCP con protocolo Streamable-HTTP

Iniciar el proyecto

Configuración del cliente

Resumen

Referencias

Planificar la estructura de directorios del proyecto (se recomienda la estructura `src/`)

Diseño principal: estructura `src/` (ventajas clave)

Usar el paquete `pyyaml` para gestionar la configuración

1. `stdio`

2. `SSE` (Server‑Sent Events, eventos enviados por el servidor)

3. `streamable‑http`

Modificar el archivo `pyproject.toml`