AMap MCP 调试 Bug 博客（EmaAgent）

更新时间：2026-02-28
环境：Windows + Conda（Ema）+ Python 3.12

背景

目标是先做一个不依赖 MCP SDK 的最小实现：

• test/amap.py 同时支持 MCP Server（--server）和小 Agent Client（默认模式）
• 通过 stdio + JSON-RPC（Content-Length 帧）通信
• Server 内部用 HTTP 调高德 Web API

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
最小可运行的 AMap MCP 示例（单文件版）

功能:
1) --server: 启动 MCP Server（stdio + Content-Length 帧）
2) 默认: 作为小 Agent（MCP Client）拉起本文件的 server 并调用工具

环境变量:
- AMAP_MAPS_API_KEY (推荐)
- AMAP_KEY (兼容)

用法示例:
1. 启动服务端（仅测试 server）:
   conda run -n Ema python test/amap.py --server

2. 作为小 Agent 调用地理编码:
   conda run -n Ema python test/amap.py geocode "北京市朝阳区望京SOHO"

3. 作为小 Agent 调用逆地理编码:
   conda run -n Ema python test/amap.py regeo "116.481488,39.990464"

4. 作为小 Agent 调用驾车路线:
   conda run -n Ema python test/amap.py route_driving "116.481488,39.990464" "116.434446,39.90816"
"""

from __future__ import annotations

import json
import os
import subprocess
import sys
from pathlib import Path
from typing import Any
from urllib.parse import urlencode
from urllib.request import Request, urlopen


AMAP_BASE_URL = "https://restapi.amap.com"


def _console_supports_utf8() -> bool:
    """
    判断当前终端是否原生支持 UTF-8 输出。
    """
    encoding = (sys.stdout.encoding or "").lower()
    return "utf" in encoding


def _dump_for_console(data: Any) -> str:
    """
    根据终端编码安全序列化 JSON 文本。

    说明:
    - 默认使用 ASCII 安全输出（\\uXXXX），规避 conda run / 控制台转码链路乱码。
    - 可通过 AMAP_JSON_MODE 控制:
      - ascii: 强制 \\uXXXX（默认，最稳）
      - auto: 根据终端编码自动判断
      - utf8: 强制中文直出
    """
    mode = (os.getenv("AMAP_JSON_MODE", "ascii") or "ascii").strip().lower()
    if mode not in {"ascii", "auto", "utf8"}:
        mode = "ascii"

    if mode == "ascii":
        return json.dumps(data, ensure_ascii=True, indent=2)
    if mode == "utf8":
        return json.dumps(data, ensure_ascii=False, indent=2)
    return json.dumps(data, ensure_ascii=not _console_supports_utf8(), indent=2)


def load_local_env() -> None:
    """
    从项目根目录的 .env 读取环境变量（仅填充未设置变量）。

    说明:
    - 不依赖 python-dotenv，减少环境前置依赖。
    - 仅支持最常见的 KEY=VALUE 行格式。
    """
    root = Path(__file__).resolve().parents[1]
    env_path = root / ".env"
    if not env_path.exists():
        return

    for raw_line in env_path.read_text(encoding="utf-8").splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, value = line.split("=", 1)
        key = key.strip()
        value = value.strip().strip("'").strip('"')
        if key and key not in os.environ:
            os.environ[key] = value


def _looks_like_placeholder_key(value: str) -> bool:
    """
    判断是否是占位用的假 key（如 YOUR_XXX_KEY）。
    """
    raw = (value or "").strip()
    if not raw:
        return True
    upper = raw.upper()
    return "YOUR_" in upper or upper.endswith("_API_KEY")


def get_amap_key() -> str:
    """
    获取高德 API Key。

    优先级:
    1) AMAP_MAPS_API_KEY
    2) AMAP_KEY
    """
    load_local_env()
    # 优先尝试 MCP 常用变量名，其次尝试兼容变量名；并跳过明显占位值。
    for env_name in ("AMAP_MAPS_API_KEY", "AMAP_KEY"):
        value = os.getenv(env_name, "").strip()
        if value and not _looks_like_placeholder_key(value):
            return value
    return ""


def http_get_json(path: str, params: dict[str, Any], timeout: int = 20) -> dict[str, Any]:
    """
    调用高德 Web API 并返回 JSON。

    Args:
        path (str): 接口路径，如 "/v3/geocode/geo"
        params (dict[str, Any]): 查询参数
        timeout (int): 超时时间（秒）

    Returns:
        dict[str, Any]: 解析后的 JSON
    """
    query = urlencode(params)
    url = f"{AMAP_BASE_URL}{path}?{query}"
    req = Request(url=url, method="GET")
    with urlopen(req, timeout=timeout) as resp:  # nosec B310
        body = resp.read().decode("utf-8", errors="replace")
        return json.loads(body)


def amap_geocode(address: str, city: str = "") -> dict[str, Any]:
    """地理编码: 地址 -> 经纬度。"""
    key = get_amap_key()
    if not key:
        raise ValueError("未找到高德 Key，请设置 AMAP_MAPS_API_KEY 或 AMAP_KEY")

    data = http_get_json(
        "/v3/geocode/geo",
        {
            "key": key,
            "address": address,
            "city": city,
            "output": "JSON",
        },
    )
    return data


def amap_regeo(location: str, radius: int = 1000) -> dict[str, Any]:
    """逆地理编码: 经纬度 -> 地址。location 格式为 'lon,lat'。"""
    key = get_amap_key()
    if not key:
        raise ValueError("未找到高德 Key，请设置 AMAP_MAPS_API_KEY 或 AMAP_KEY")

    data = http_get_json(
        "/v3/geocode/regeo",
        {
            "key": key,
            "location": location,
            "radius": radius,
            "extensions": "all",
            "output": "JSON",
        },
    )
    return data


def amap_route_driving(origin: str, destination: str, strategy: int = 0) -> dict[str, Any]:
    """驾车路径规划。origin/destination 格式为 'lon,lat'。"""
    key = get_amap_key()
    if not key:
        raise ValueError("未找到高德 Key，请设置 AMAP_MAPS_API_KEY 或 AMAP_KEY")

    data = http_get_json(
        "/v3/direction/driving",
        {
            "key": key,
            "origin": origin,
            "destination": destination,
            "strategy": strategy,
            "extensions": "base",
            "output": "JSON",
        },
    )
    return data


def _read_message(stdin_buffer: Any) -> dict[str, Any] | None:
    """
    读取一条 MCP framed message（Content-Length + JSON body）。

    协议格式示例:
      Content-Length: 123\r\n
      Content-Type: application/json\r\n
      \r\n
      { ...json... }
    """
    headers: dict[str, str] = {}
    while True:
        line = stdin_buffer.readline()
        if not line:
            return None
        if line in (b"\r\n", b"\n"):
            break
        text = line.decode("utf-8", errors="replace").strip()
        if ":" in text:
            k, v = text.split(":", 1)
            headers[k.strip().lower()] = v.strip()

    content_length = int(headers.get("content-length", "0"))
    if content_length <= 0:
        return None

    body = stdin_buffer.read(content_length)
    if not body:
        return None
    return json.loads(body.decode("utf-8", errors="replace"))


def _write_message(stdout_buffer: Any, payload: dict[str, Any]) -> None:
    """写出一条 MCP framed message。"""
    # MCP 传输层固定写 ASCII JSON（中文转 \uXXXX），最大程度规避环境编码差异。
    body = json.dumps(payload, ensure_ascii=True).encode("utf-8")
    header = f"Content-Length: {len(body)}\r\nContent-Type: application/json\r\n\r\n".encode("utf-8")
    stdout_buffer.write(header)
    stdout_buffer.write(body)
    stdout_buffer.flush()


def _tool_schema() -> list[dict[str, Any]]:
    """返回 MCP tools/list 所需工具定义。"""
    return [
        {
            "name": "maps_geocode",
            "description": "根据结构化地址查询经纬度",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "address": {"type": "string", "description": "详细地址"},
                    "city": {"type": "string", "description": "城市，可选"},
                },
                "required": ["address"],
            },
        },
        {
            "name": "maps_regeo",
            "description": "根据经纬度反查地址信息",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "location": {"type": "string", "description": "格式: lon,lat"},
                    "radius": {"type": "integer", "description": "搜索半径，默认1000"},
                },
                "required": ["location"],
            },
        },
        {
            "name": "maps_route_driving",
            "description": "驾车路线规划",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "origin": {"type": "string", "description": "起点，格式: lon,lat"},
                    "destination": {"type": "string", "description": "终点，格式: lon,lat"},
                    "strategy": {"type": "integer", "description": "规划策略，默认0"},
                },
                "required": ["origin", "destination"],
            },
        },
    ]


def _handle_tool_call(name: str, arguments: dict[str, Any]) -> dict[str, Any]:
    """分发并执行具体工具。"""
    if name == "maps_geocode":
        address = str(arguments.get("address", "")).strip()
        city = str(arguments.get("city", "")).strip()
        if not address:
            raise ValueError("address 不能为空")
        result = amap_geocode(address=address, city=city)
    elif name == "maps_regeo":
        location = str(arguments.get("location", "")).strip()
        radius = int(arguments.get("radius", 1000))
        if not location:
            raise ValueError("location 不能为空")
        result = amap_regeo(location=location, radius=radius)
    elif name == "maps_route_driving":
        origin = str(arguments.get("origin", "")).strip()
        destination = str(arguments.get("destination", "")).strip()
        strategy = int(arguments.get("strategy", 0))
        if not origin or not destination:
            raise ValueError("origin/destination 不能为空")
        result = amap_route_driving(origin=origin, destination=destination, strategy=strategy)
    else:
        raise ValueError(f"未知工具: {name}")

    return {
        "content": [
            {
                "type": "text",
                "text": _dump_for_console(result),
            }
        ],
        "isError": False,
    }


def run_mcp_server() -> None:
    """启动最小 MCP Server（stdio 传输）。"""
    stdin_buffer = sys.stdin.buffer
    stdout_buffer = sys.stdout.buffer

    while True:
        req = _read_message(stdin_buffer)
        if req is None:
            break

        req_id = req.get("id")
        method = req.get("method", "")
        params = req.get("params") or {}

        # Notification 无需响应
        if req_id is None:
            continue

        try:
            if method == "initialize":
                result = {
                    "protocolVersion": "2024-11-05",
                    "serverInfo": {"name": "amap-mcp-local", "version": "0.1.0"},
                    "capabilities": {"tools": {}},
                }
            elif method == "tools/list":
                result = {"tools": _tool_schema()}
            elif method == "tools/call":
                tool_name = str(params.get("name", "")).strip()
                arguments = params.get("arguments") or {}
                if not isinstance(arguments, dict):
                    raise ValueError("arguments 必须是 object")
                result = _handle_tool_call(tool_name, arguments)
            else:
                raise ValueError(f"不支持的方法: {method}")

            resp = {"jsonrpc": "2.0", "id": req_id, "result": result}
        except Exception as exc:
            resp = {
                "jsonrpc": "2.0",
                "id": req_id,
                "error": {"code": -32000, "message": str(exc)},
            }

        _write_message(stdout_buffer, resp)


class TinyMCPAgent:
    """
    一个最小 Agent：通过 stdio 拉起本地 MCP Server 并调用工具。

    说明:
    - 仅用于本地验证 MCP 协议流程，不依赖第三方 MCP SDK。
    """

    def __init__(self) -> None:
        self._next_id = 1
        self.proc: subprocess.Popen[bytes] | None = None

    def start(self) -> None:
        """启动本文件的 --server 子进程。"""
        script_path = str(Path(__file__).resolve())
        self.proc = subprocess.Popen(
            [sys.executable, script_path, "--server"],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            cwd=str(Path(__file__).resolve().parent),
        )

    def stop(self) -> None:
        """停止子进程。"""
        if self.proc and self.proc.poll() is None:
            self.proc.terminate()
            try:
                self.proc.wait(timeout=3)
            except Exception:
                self.proc.kill()
        self.proc = None

    def _send(self, payload: dict[str, Any]) -> None:
        if not self.proc or not self.proc.stdin:
            raise RuntimeError("MCP server 未启动")
        _write_message(self.proc.stdin, payload)

    def _recv(self) -> dict[str, Any]:
        if not self.proc or not self.proc.stdout:
            raise RuntimeError("MCP server 未启动")
        msg = _read_message(self.proc.stdout)
        if msg is None:
            raise RuntimeError("MCP server 无响应")
        return msg

    def request(self, method: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
        """
        发送 JSON-RPC 请求并等待返回。

        Args:
            method (str): RPC 方法名
            params (dict[str, Any] | None): 参数

        Returns:
            dict[str, Any]: result 字段内容
        """
        req_id = self._next_id
        self._next_id += 1
        req = {"jsonrpc": "2.0", "id": req_id, "method": method, "params": params or {}}
        self._send(req)

        while True:
            resp = self._recv()
            if resp.get("id") != req_id:
                continue
            if "error" in resp:
                raise RuntimeError(resp["error"].get("message", "unknown rpc error"))
            return resp.get("result", {})

    def notify(self, method: str, params: dict[str, Any] | None = None) -> None:
        """发送 notification（无 id，无响应）。"""
        self._send({"jsonrpc": "2.0", "method": method, "params": params or {}})

    def initialize(self) -> dict[str, Any]:
        """执行 initialize + initialized 通知。"""
        result = self.request(
            "initialize",
            {
                "protocolVersion": "2024-11-05",
                "clientInfo": {"name": "tiny-amap-agent", "version": "0.1.0"},
                "capabilities": {},
            },
        )
        self.notify("notifications/initialized", {})
        return result

    def list_tools(self) -> list[dict[str, Any]]:
        """读取 MCP tools/list。"""
        result = self.request("tools/list", {})
        return list(result.get("tools") or [])

    def call_tool(self, name: str, arguments: dict[str, Any]) -> dict[str, Any]:
        """调用 MCP tools/call。"""
        return self.request("tools/call", {"name": name, "arguments": arguments})


def run_agent_cli(argv: list[str]) -> int:
    """
    小 Agent 命令行入口。

    支持:
    - geocode <address> [city]
    - regeo <lon,lat> [radius]
    - route_driving <origin_lon,lat> <dest_lon,lat> [strategy]
    """
    if len(argv) < 2:
        print("用法:")
        print("  python test/amap.py geocode <address> [city]")
        print("  python test/amap.py regeo <lon,lat> [radius]")
        print("  python test/amap.py route_driving <origin> <destination> [strategy]")
        return 1

    cmd = argv[1].strip().lower()
    agent = TinyMCPAgent()
    agent.start()
    try:
        init_result = agent.initialize()
        print("[MCP initialize]")
        print(_dump_for_console(init_result))

        tools = agent.list_tools()
        print("\n[MCP tools/list]")
        print(_dump_for_console(tools))

        if cmd == "geocode":
            address = argv[2] if len(argv) > 2 else ""
            city = argv[3] if len(argv) > 3 else ""
            result = agent.call_tool("maps_geocode", {"address": address, "city": city})
        elif cmd == "regeo":
            location = argv[2] if len(argv) > 2 else ""
            radius = int(argv[3]) if len(argv) > 3 else 1000
            result = agent.call_tool("maps_regeo", {"location": location, "radius": radius})
        elif cmd == "route_driving":
            origin = argv[2] if len(argv) > 2 else ""
            destination = argv[3] if len(argv) > 3 else ""
            strategy = int(argv[4]) if len(argv) > 4 else 0
            result = agent.call_tool(
                "maps_route_driving",
                {"origin": origin, "destination": destination, "strategy": strategy},
            )
        else:
            raise ValueError(f"未知命令: {cmd}")

        print("\n[MCP tools/call result]")
        print(_dump_for_console(result))
        return 0
    finally:
        agent.stop()


if __name__ == "__main__":
    if "--server" in sys.argv:
        run_mcp_server()
    else:
        raise SystemExit(run_agent_cli(sys.argv))

---

遇到的主要报错

1. INVALID_USER_KEY / 10001

典型返回：

{"status":"0","info":"INVALID_USER_KEY","infocode":"10001"}

现象：有时同一机器上又能跑出 status=1。

根因分析：

• .env 里可能同时存在真实 Key 和占位 Key（如 YOUR_XXX_KEY）。
• 代码读取优先级不当时，会优先拿到占位值。

处理：

• 读取 Key 时过滤占位形式（如 YOUR_、*_API_KEY 占位名）。
• 统一 Key 来源优先级，避免“表面有值，实际无效”。

---

2. 子进程路径错误：test\\test\\amap.py not found

报错示例：

python: can't open file './EmaAgent\\test\\test\\amap.py': [Errno 2] No such file or directory

根因分析：

• 在 cwd=test 的前提下，子进程又使用相对路径 test/amap.py，导致路径重复拼接。

处理：

• 子进程启动脚本使用绝对路径 Path(__file__).resolve()。

---

3. 中文乱码（鏍规嵁... / 鍖椾含...）

现象：

• MCP 返回内容、工具描述出现乱码。
• 同样代码，有时直接 python 正常，有时 conda run 乱码。

关键结论：

• 业务数据本身没错，乱码主要发生在“输出链路转码”。

补充验证：

• 直接运行（conda activate Ema 后 python test/amap.py ...）可正常。
• conda run -n Ema python ... 更容易乱码。

根因分析：

• conda run 会包一层执行并捕获子进程输出，再转发到当前终端。
• 这层可能按 GBK/CP936 处理字节流，而程序/数据是 UTF-8，导致“UTF-8 被当 GBK 解码”。

处理：

• 输出策略增加安全模式：默认 ASCII JSON（中文转 \uXXXX），确保不乱码。
• 需要中文直出时再手动切到 UTF-8 模式。

---

4. Conda 自身输出异常（UnicodeEncodeError）

出现过类似：

UnicodeEncodeError: 'gbk' codec can't encode character ...

根因分析：

• 不是业务逻辑报错，而是 Conda 包装输出阶段编码不一致导致。

处理：

• 调试时优先用目标环境 python.exe 直接跑，绕过 conda run 包装层。

---

哪些地方其实没问题（容易被误判）

http_get_json(...) 逻辑是正确的

def http_get_json(path: str, params: dict[str, Any], timeout: int = 20) -> dict[str, Any]:
    query = urlencode(params)
    url = f"{AMAP_BASE_URL}{path}?{query}"
    req = Request(url=url, method="GET")
    with urlopen(req, timeout=timeout) as resp:
        body = resp.read().decode("utf-8", errors="replace")
        return json.loads(body)

说明：

• 高德 Web API 返回 JSON，utf-8 decode + json.loads 是标准做法。
• 如果这里有问题，通常会是 JSON 解析失败，不会只表现为“中文展示乱码”。

MCP 写帧函数本质也没问题

• 核心是 Content-Length + body 的 framing。
• 本次增强是把 JSON 设为 ASCII 安全输出，提升跨终端鲁棒性，不是协议修错。

---

最终策略（落地）

1. MCP 传输层尽量使用 ASCII 安全 JSON（\uXXXX），规避终端/中间层转码污染。
2. 本地调试优先：
   • conda activate Ema
   • python test/amap.py ...
3. Key 读取逻辑必须防占位值污染。
4. 子进程脚本路径一律绝对路径。
5. 日志和展示层与业务逻辑分离看待：
   • 数据正确 != 显示一定正确（编码链路可能出问题）。

---

复盘反思

反思 1：先区分“数据错”还是“显示错”

一开始把乱码当成 API 返回异常，排查方向偏慢。
后续通过 repr、直连 python.exe、比对 conda run 与 activate 结果，才确认是输出链路问题。

反思 2：Windows + Conda + 多层包装时，编码要先做稳态设计

不是“只要 UTF-8 就万事大吉”。
中间层（如 conda run）会引入二次解码风险，必须考虑兼容策略（ASCII 安全输出）。

反思 3：MCP 最小可用优先

先保证协议流程跑通（initialize/list/call），再追求漂亮中文展示。
这次先稳定了可用性，后续再优化 UX（例如前端渲染时统一解码展示）。

---