vent_deep_agent/list_mcp_tools.py

# -*- coding: utf-8 -*-
"""
MCP 远程工具列表查询 —— 格式化输出所有可用的 MCP 工具及参数签名
"""

import asyncio
import shutil
from fastmcp import Client

# ── 终端宽度 ──
WIDTH = min(shutil.get_terminal_size().columns, 120)

# ── 工具分类定义（按名称前缀归类）──
CATEGORY_MAP = {
    "vent":   "🌬️  通风监测",
    "device": "🔧 设备管理",
    "tunnel": "🛤️  巷道信息",
    "mine":   "⛏️  煤矿基础",
    "needq":  "📊 需风量计算",
}

def _category(name: str) -> str:
    for prefix, label in CATEGORY_MAP.items():
        if name.startswith(prefix):
            return label
    return "📦 其他"


def parse_tool_params(schema: dict) -> list[dict]:
    """解析 inputSchema 为参数列表"""
    params = []
    required = schema.get("required", [])
    props = schema.get("properties", {})
    for name, info in props.items():
        params.append({
            "name": name,
            "type": info.get("type", "—"),
            "desc": (info.get("description", "") or "")[:120],
            "required": name in required,
        })
    return params


# ── 主入口 ──

MCP_URL = "http://39.97.59.228:8071/mcp"
client = Client(MCP_URL)


async def list_all_mcp_tools_tcp():
    """TCP 远程 MCP 客户端查询工具列表"""
    async with client:
        tools = await client.list_tools()

        # ══════════════════════════════════════════════
        #  头部
        # ══════════════════════════════════════════════
        print()
        print("╔" + "═" * (WIDTH - 2) + "╗")
        print("║" + f"  🔌  MCP 远程工具列表  |  {MCP_URL}  |  共 {len(tools)} 个工具".center(WIDTH - 2) + "║")
        print("╚" + "═" * (WIDTH - 2) + "╝")
        print()

        # 按类别分组
        groups: dict[str, list] = {}
        for t in tools:
            cat = _category(t.name)
            groups.setdefault(cat, []).append(t)

        order = ["🌬️  通风监测", "🛤️  巷道信息", "🔧 设备管理", "📊 需风量计算", "⛏️  煤矿基础", "📦 其他"]
        index = 0

        for cat in order:
            if cat not in groups:
                continue
            group_tools = groups[cat]
            print(f"  {cat}（{len(group_tools)} 个）")
            print(f"  {'─' * (WIDTH - 6)}")

            for t in group_tools:
                index += 1
                # ── 工具名 + 序号 ──
                print(f"\n  ┌─ [{index:02d}] {t.name} {'─' * max(WIDTH - len(t.name) - 15, 0)}")

                # ── 描述 ──
                desc = (t.description or "").strip()
                if desc:
                    # 自动换行
                    for line in _wrap(desc, WIDTH - 8):
                        print(f"  │  📝 {line}")

                # ── 参数表 ──
                params = parse_tool_params(t.inputSchema)
                if params:
                    print(f"  │  ┌ 参数{'─' * (WIDTH - 13)}")
                    for p in params:
                        flag = "🔴" if p["required"] else "🟢"
                        print(f"  │  │ {flag} {p['name']}: {p['type']}")
                        if p["desc"]:
                            for line in _wrap(p["desc"], WIDTH - 12):
                                print(f"  │  │    {line}")
                    print(f"  │  └{'─' * (WIDTH - 9)}")
                else:
                    print(f"  │  (无参数)")

                print(f"  └{'─' * (WIDTH - 6)}")

            print()

        # ══════════════════════════════════════════════
        #  底部统计
        # ══════════════════════════════════════════════
        total_params = sum(
            len(parse_tool_params(t.inputSchema)) for t in tools
        )
        print("╔" + "═" * (WIDTH - 2) + "╗")
        print("║" + f"  ✅ 总计 {len(tools)} 个工具，{total_params} 个参数，{len(groups)} 个分类".ljust(WIDTH - 2) + "║")
        print("╚" + "═" * (WIDTH - 2) + "╝")
        print()


def _wrap(text: str, width: int) -> list[str]:
    """按宽度自动换行，尊重中文宽度"""
    lines = []
    while len(text) > width:
        # 尝试在空格处断行
        cut = text.rfind(" ", 0, width)
        if cut == -1:
            cut = width
        lines.append(text[:cut].strip())
        text = text[cut:].strip()
    if text:
        lines.append(text)
    return lines


asyncio.run(list_all_mcp_tools_tcp())