Gaode Map POI

Fornece informações de geolocalização e POI (Pontos de Interesse) próximos usando a API do Gaode Map.

Documentação

Tutorial MCP nível babá: do zero até a implantação, tudo em um!

O quanto o MCP está em alta dispensa repetições. Como uma tecnologia emergente, há uma infinidade de materiais na internet chinesa sobre como desenvolver serviços MCP, mas a maioria é vaga ou superficial, e a maioria dos exemplos apenas copia os exemplos da documentação oficial em artigos rasos.

Como desenvolvedor, sei bem a importância da engenharia. O desenvolvimento de um serviço MCP não é apenas escrever duas interfaces de serviço imitando exemplos; é preciso considerar estrutura de código, gerenciamento de configuração, registro de logs, tratamento de exceções e muito mais.

Após exploração e prática, organizei o fluxo de desenvolvimento de um serviço MCP em um guia detalhado, na esperança de ajudar mais desenvolvedores a começar rapidamente e construir serviços MCP de alta qualidade.

O código deste projeto é open source no GitHub, e todos são bem-vindos para dar Star e Fork.

Que tipo de serviço MCP vamos construir?

Para termos um objetivo prático melhor, vamos construir um serviço MCP baseado na API do Gaode Map, com as seguintes funcionalidades principais:

Obter a localização geográfica do usuário com base no IP

Parâmetros: ip地址(可选，不传递默认获取当前主机IP)
Retorno: informações de latitude e longitude do usuário

Obter informações de POI próximos com base na localização geográfica do usuário

Parâmetros: 经度、纬度、POI类型
Retorno: lista de informações de POI próximos

Após concluir as funcionalidades acima, podemos obter informações reais de POI por meio de conversas com o modelo de linguagem grande. Por exemplo:

Usuário: Quero saber quais restaurantes estão perto de mim?
Serviço MCP: Com base na sua localização, há os seguintes restaurantes próximos: 1. Restaurante A 2. Restaurante B 3. Restaurante C
Usuário: Qual é o endereço do Restaurante A?
Serviço MCP: O endereço do Restaurante A é:
- Endereço: XXX
- Telefone: XXX
- Horário de funcionamento: XXX

Exemplo visual

Conceitos principais do MCP

Se você não tem ideia do que é MCP antes de começar, recomendo ler a documentação oficial do MCP para entender os conceitos básicos.

Um servidor MCP pode fornecer três tipos principais de funcionalidades:

Resources: recursos, dados semelhantes a arquivos que o cliente pode ler (como respostas de API ou conteúdo de arquivos)
Tools: ferramentas, funções que o LLM pode chamar (com aprovação do usuário)
Prompts: prompts, modelos pré-escritos que ajudam o usuário a concluir tarefas específicas

Conhecimentos necessários

Antes de começar, recomenda-se ter os seguintes conhecimentos:

Noções básicas de Python
Conceitos de LLM (Large Language Model)
UV (ferramenta de gerenciamento de pacotes Python)

Requisitos do sistema

Python 3.10 ou superior
Python MCP SDK 1.2.0

Configurando o ambiente de desenvolvimento

⚠ Certifique-se de ajustar os comandos de acordo com seu sistema operacional; a sintaxe de comandos do powershell e do bash é diferente.

O autor usa Windows + terminal Git. A primeira metade deste tutorial é basicamente igual à oficial; consulte o exemplo de desenvolvimento de servidor na documentação oficial.

Instalando o 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"

Criando ambiente virtual e inicializando o projeto

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

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

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

Planejando a estrutura de diretórios do projeto (recomendado layout `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álise do design da estrutura

Design principal: layout `src/` (vantagens-chave)

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

Por que usar essa estrutura?

✅ Isolamento do ambiente de instalação (valor central)
Durante os testes, força a instalação do pacote via pip install, evitando referência direta ao caminho do código-fonte, garantindo que o ambiente de teste = ambiente de execução do usuário
✅ Previne dependência implícita de caminhos
Elimina importações incorretas causadas pelo diretório de desenvolvimento estar em primeiro lugar em sys.path (comum em layouts tradicionais sem src/)
✅ Segurança de empacotamento
Força a validação de que o conteúdo do pacote está corretamente incluído nos arquivos de distribuição (arquivos ausentes são imediatamente expostos nos testes)
✅ Consistência entre ambientes
Ambientes de desenvolvimento/teste/produção usam exatamente a mesma estrutura de pacote, eliminando o problema de "funciona na minha máquina"

📊 Dados: pesquisa oficial da PyPA mostra que projetos com layout src/ têm taxa de erro de empacotamento 63% menor (fonte)

Escrevendo o código das ferramentas

Após planejar os diretórios, começamos a codificar de fato. Um projeto formal e padronizado pode envolver muitas leituras de configuração. Primeiro, encapsulamos a funcionalidade de leitura de arquivos de configuração.

Usando o pacote `pyyaml` para gerenciar configurações

uv add pyyaml

Criando o módulo de gerenciamento de configuração

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

Criando o arquivo de configuração

touch src/build_mcp/config.yaml

Adicione o seguinte conteúdo ao arquivo 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

⚠ O arquivo config.yaml precisa estar no diretório src/build_mcp/, para que seja encontrado corretamente ao carregar a configuração.

Este arquivo de configuração serve apenas como exemplo de engenharia; em ambientes de produção, não escreva informações sensíveis (como chaves de API) diretamente no arquivo de configuração. Recomenda-se usar variáveis de ambiente ou serviços de armazenamento seguro.

Escrevendo o código de gerenciamento de configuração

# 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

Instalando o código

⚠ Ao instalar o código pela primeira vez, use o comando pip install -e ., para que o diretório atual seja instalado como um pacote editável no ambiente virtual. Assim, alterações no código durante o desenvolvimento entram em vigor imediatamente, sem necessidade de reinstalação.

uv pip install -e .

Escrevendo código de teste

Escrever código de teste o mais detalhado possível no projeto é um bom hábito. Na engenharia de projetos, escrevemos testes para funcionalidades principais, garantindo a correção e estabilidade do 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"

Executando os testes

uv run pytest tests

Escrevendo o módulo de logs

Como programador, a capacidade de localizar problemas rapidamente é muito importante, e o sistema de logs é essencial. Um bom programador não só sabe escrever código, mas também sabe escrever logs. Vamos encapsular um módulo de logs simples para uso posterior.

touch src/build_mcp/common/logger.py

Aqui implementamos um sistema de logs que gera saída tanto no console quanto em arquivo, com suporte a rotação e backup 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

Até agora, os módulos básicos do sistema foram construídos. Em seguida, implementaremos as funcionalidades principais do serviço.

Escrevendo o SDK de requisições do Gaode Map

De acordo com a documentação da API do Gaode Map, precisamos implementar duas funcionalidades principais:

Obter localização geográfica com base no IP do usuário
Obter informações de POI próximos com base na localização geográfica

Criando o módulo de serviço do Gaode Map

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

Escrevendo o código do serviço do 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

O código implementa:

Requisições HTTP assíncronas, com suporte a nova tentativa automática e backoff exponencial
Método locate_ip para obter localização geográfica com base no IP
Método de busca por proximidade search_nearby, para obter informações de POI próximos com base em latitude e longitude

Escrevendo código de teste para o serviço do 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"

Executando os testes

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

🚀 Introdução aos três protocolos de transporte do MCP

1. `stdio`

Método de comunicação: entre processos locais, mensagens JSON‑RPC são transmitidas bidirecionalmente via entrada/saída padrão (stdin/stdout);
Cenário de uso: chamada local de ferramentas ou subprocessos, como integração leve em aplicativos de desktop;
Vantagens: baixa latência, implementação simples, sem necessidade de rede.

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

Método de comunicação: baseado em HTTP: o cliente envia mensagens via POST, e o servidor estabelece push unidirecional text/event‑stream via GET;
Status atual: está obsoleto (deprecated), substituído por "streamable‑http" desde MCP v2024‑11‑05, mas ainda mantém suporte de compatibilidade;
Vantagens: adequado para implementações simples em cenários remotos iniciais que exigem apenas push do servidor;
Desvantagens: apenas unidirecional servidor→cliente, e a conexão não suporta retomada de ponto de interrupção.

3. `streamable‑http`

Método de comunicação: transporte bidirecional baseado em HTTP: o cliente faz requisições JSON‑RPC via POST, e o servidor pode retornar respostas únicas (JSON) ou mensagens SSE em streaming, além de estabelecer push do servidor via GET;
Funcionalidades suportadas:
- Um único endpoint /mcp lida com toda a comunicação;
- Gerenciamento de sessão (via Mcp‑Session‑Id);
- Retomada de fluxo e reprodução de mensagens (recuperação de desconexão HTTP suporta Last‑Event‑ID);
- Compatibilidade retroativa com SSE;
Status atual: recomendado por padrão desde MCP v2025‑03‑26, adequado para implantação em nuvem e remota, sendo a escolha preferida para cenários remotos; (modelcontextprotocol.io)

📊 Visão geral da comparação de protocolos

Protocolo	Direção de comunicação	Cenário de uso	Destaques	Nível de recomendação
stdio	Bidirecional (local)	Chamada de subprocesso local	Simples, baixa latência, zero dependência de rede	⭐ Preferido local
SSE	Unidirecional (servidor→cliente)	Implementação remota inicial	Implementação simples, mas sem suporte a retomada	⚠️ Obsoleto
streamable‑http	Bidirecional/push SSE opcional	Interação em nuvem/remota	Endpoint único, multifuncional, retomada de fluxo, alta compatibilidade	✅ Recomendado

Escrevendo o programa principal do serviço MCP

Em seguida, escrevemos o programa principal do serviço MCP, que processa requisições do cliente e chama o SDK do 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))

No código, encapsulamos uma classe de resposta unificada e fornecemos duas funções de ferramenta:

locate_ip: obtém a localização geográfica com base no endereço IP
search_nearby: realiza busca por proximidade com base em latitude, longitude e palavra-chave

É importante notar que o tipo Annotated no código é essencial, pois permite que o LLM chame as ferramentas com mais precisão por meio de metainformações. Atualmente, a maioria dos desenvolvedores de serviços MCP não tem essa consciência, apenas definem as ferramentas de forma simples, o que resulta em um desempenho muito ruim.

Também escrevemos um prompt, que é fornecido no contexto da conversa. Esse é um ponto muito importante que muitos desenvolvedores não percebem. Na era da IA, não basta escrever bom código; precisamos aprender a refinar os prompts.

Na verdade, o núcleo principal do artigo está nessa parte do código; por favor, entenda essas informações com atenção.

Neste ponto, concluímos a implementação das funcionalidades principais do serviço MCP. Em seguida, precisamos escrever o ponto de entrada do serviço e iniciar o serviço MCP.

Escrevendo o ponto de entrada do serviço 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()

Modificando o arquivo `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"]

O ponto principal é

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

Isso significa:

Ao executar o comando build_mcp, será equivalente a executar:

from build_mcp import main
main()

Podemos executar o serviço MCP com os seguintes comandos:

Iniciar o serviço MCP com protocolo stdio:

uv run build_mcp

Iniciar o serviço MCP com protocolo streamable-http:

uv run build_mcp streamable-http

Depurando o serviço MCP

Como depurar o serviço MCP depende da forma como iniciamos o serviço.

1. Escrevendo código de cliente para depurar o serviço MCP com 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)

Executando os testes

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

Este é um código de teste; aqui é apenas um exemplo. Você pode escrever mais código de teste de acordo com suas necessidades para validar as funcionalidades do serviço MCP.

2. Testando com o Inspector

O Inspector é uma ferramenta de depuração oficial para serviços MCP. Com ela, você pode iniciar uma interface web local e chamar diretamente as ferramentas do serviço MCP na interface. É mais intuitiva e fácil de usar; essa abordagem é recomendada. Para mais detalhes, consulte a documentação oficial.

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

Escrevendo o Makefile

Para facilitar o desenvolvimento e os testes, podemos escrever um Makefile para gerenciar comandos comuns.

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

Até agora, desenvolvemos completamente um serviço MCP do zero ao fim. Parabéns a nós por aprendermos uma nova habilidade!

Como usar este serviço MCP?

Primeiro, você precisa ter um cliente MCP. Atualmente, há vários tipos de clientes MCP no mercado; qual usar fica a seu critério.

Aqui está um guia muito detalhado de uso de clientes MCP, um ótimo projeto no GitHub: Guia de uso de clientes MCP

Escolha um cliente, baixe e instale, e então configure o serviço que desenvolvemos.

Configurando o serviço MCP com protocolo Stdio

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

⚠ Preste atenção ao ambiente UV local; se vários UVs estiverem instalados, pode causar confusão no ambiente. Esse é um ponto problemático no desenvolvimento; fique atento.

Configurando o serviço MCP com protocolo Streamable-HTTP

Iniciando o projeto

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)

Após a inicialização bem-sucedida, um serviço HTTP será iniciado na porta 8000.

Configuração do cliente

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

Resumo

Este artigo apresentou como construir um serviço MCP do Gaode Map do zero, cobrindo o seguinte:

Conceitos básicos e configuração de serviços MCP
Como usar a API do Gaode Map para localização por IP e busca por proximidade
Como escrever as funcionalidades principais do serviço MCP, incluindo gerenciamento de configuração, sistema de logs e SDK do Gaode Map
Como escrever o programa principal e o ponto de entrada do serviço MCP
Como depurar o serviço MCP, incluindo o uso do Inspector e a escrita de código de teste
Como usar o Makefile para gerenciar comandos do projeto
Como configurar um cliente MCP para conectar ao nosso serviço

O artigo termina aqui. Desejamos a todos bons estudos! Se você tiver alguma dúvida ou sugestão, envie uma issue ou pull request para o repositório do GitHub

Gaode Map POI

Documentação

Tutorial MCP nível babá: do zero até a implantação, tudo em um!

O código deste projeto é open source no GitHub, e todos são bem-vindos para dar Star e Fork.

Que tipo de serviço MCP vamos construir?

Obter a localização geográfica do usuário com base no IP

Obter informações de POI próximos com base na localização geográfica do usuário

Exemplo visual

Conceitos principais do MCP

Conhecimentos necessários

Requisitos do sistema

Configurando o ambiente de desenvolvimento

Instalando o UV

Criando ambiente virtual e inicializando o projeto

Planejando a estrutura de diretórios do projeto (recomendado layout src/)

Análise do design da estrutura

Design principal: layout src/ (vantagens-chave)

Escrevendo o código das ferramentas

Usando o pacote pyyaml para gerenciar configurações

Criando o módulo de gerenciamento de configuração

Criando o arquivo de configuração

Escrevendo o código de gerenciamento de configuração

Instalando o código

Escrevendo código de teste

Executando os testes

Escrevendo o módulo de logs

Escrevendo o SDK de requisições do Gaode Map

Criando o módulo de serviço do Gaode Map

Escrevendo o código do serviço do Gaode Map

Escrevendo código de teste para o serviço do Gaode Map

Executando os testes

🚀 Introdução aos três protocolos de transporte do MCP

1. stdio

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

3. streamable‑http

📊 Visão geral da comparação de protocolos

Escrevendo o programa principal do serviço MCP

Escrevendo o ponto de entrada do serviço MCP

Modificando o arquivo pyproject.toml

Depurando o serviço MCP

1. Escrevendo código de cliente para depurar o serviço MCP com protocolo stdio

Executando os testes

2. Testando com o Inspector

Escrevendo o Makefile

Como usar este serviço MCP?

Configurando o serviço MCP com protocolo Stdio

Configurando o serviço MCP com protocolo Streamable-HTTP

Iniciando o projeto

Configuração do cliente

Resumo

Referências

Planejando a estrutura de diretórios do projeto (recomendado layout `src/`)

Design principal: layout `src/` (vantagens-chave)

Usando o pacote `pyyaml` para gerenciar configurações

1. `stdio`

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

3. `streamable‑http`

Modificando o arquivo `pyproject.toml`