chendandan_homework/scripts/test_amap_mcp.py

from __future__ import annotations
"""测试高德MCP的文本搜索与周边搜索等工具调用。"""
import asyncio
import json
from typing import Any

# 直接运行本文件时(如 python scripts/test_amap_mcp.py),Python 只把 scripts/ 加入 sys.path,
# 找不到项目根的 app 包。此时需要把项目根目录注入 sys.path:
# sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
# 使用 `python -m scripts.test_amap_mcp` 运行时则无需此处理,项目根会自动加入 sys.path。
from rich.console import Console
from rich.table import Table
from langchain_core.tools import BaseTool
from langchain_mcp_adapters.client import MultiServerMCPClient
from app.config import get_settings

console = Console()

def get_tool_schema(tool:BaseTool)->dict[str, Any]:
    """兼容不同 LangChain 工具 Schema 表示。"""

    schema = tool.args_schema

    if schema is None:
        return {}
    if isinstance(schema,dict):
        return schema
    if hasattr(schema,"model_json_schema"):
        return schema.model_json_schema()
    return {}

async def invoke_text_search(tools:list[BaseTool])->None:
    """如果发现高德关键词搜索工具，则执行一次真实POI查询。"""
    candidate_names = {
        "maps_text_search",
        "map_text_search",
        "text_search",
    }

    search_tool = next(
        (
            tool
            for tool in tools
            if tool.name in candidate_names
            or "text_search" in tool.name.lower()
        ),
        None,
    )

    if search_tool is None:
        console.print(
            "[yellow]MCP 已连接成功，但没有匹配到预设的关键词搜索工具名。[/yellow]"
        )
        console.print(
            "这是正常的版本差异。先保留工具列表，后续根据实际名称接入。"
        )
        return
    
    schema = get_tool_schema(search_tool)
    properties = schema.get("properties",{})
    # 注意:JSON Schema 中 required 和 properties 是平级字段,不是嵌套关系。
    required = schema.get("required",[])

    console.rule("[bold]关键词搜索工具 Schema")
    console.print_json(json.dumps(schema,ensure_ascii=False))

    args:dict[str, Any] = {}

    keyword_names = {"keywords","keyword","query"}
    city_names = {"city","region"}

    
    for name in properties:
        if name in keyword_names:
            args[name] = "宽窄巷子"
        elif name in city_names:
            args[name] = "成都"


    missing_required = [
        name
        for name in required
        if name not in args
    ]

    if missing_required:
        console.print(
            "[yellow]发现关键词搜索工具，但存在无法自动填充的必填参数："
            f"{missing_required}[/yellow]"
        )
        return
    
    console.rule("[bold]执行真实 POI 查询")
    console.print(f"工具：{search_tool.name}")
    console.print(f"参数：{args}")

    result = await search_tool.ainvoke(args)

    console.print("[green]POI 查询成功：[/green]")
    console.print(result)

async def main()->None:
    """主函数，用于测试高德关键词搜索工具。"""
    settings = get_settings()
    # 高德 key 在 modelscope MCP 端点中通常通过 URL token 鉴权,这里读取备用。
    settings.require("amap_api_key")

    mcp_client = MultiServerMCPClient(
        {
            "amap": {
                "transport": "streamable_http",
                "url": "https://mcp.api-inference.modelscope.net/8ac378c1961447/mcp",
            }
        }
    )

    console.rule("[bold]连接 高德MCP")

    tools = await mcp_client.get_tools()

    if not tools:
        raise RuntimeError("MCP 连接成功，但服务端未返回任何工具")
    
    table = Table(title=f"高德MCP 工具列表,共{len(tools)}个")
    table.add_column("序号")
    table.add_column("工具名")
    table.add_column("说明")

    for index,tool in enumerate(tools,start=1):
        description = (tool.description or "").replace("\n"," ")

        table.add_row(
            str(index),
            tool.name,
            description[:100],
        )
    
    console.print(table)

    await invoke_text_search(tools)

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except Exception as e:
        console.print(f"[red]高德mcp测试失败：{type(e).__name__}:{e}[/red]")
        raise SystemExit(1) from e