vent_deep_agent/xfl_demo_client.py

import asyncio
import json

from fastmcp import Client

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

async def call_tool():
    async with client:
        result1 = await client.call_tool("query_device_data_by_id",{"device_id": "1432344790415630337", "page_size": 20})
        data = json.loads(result1.content[0].text)
        print(json.dumps(data, ensure_ascii=False, indent=2))

        # result2 = await client.call_tool("list_ventanaly_monitor_data_days",{"strtype":"1730768271306147"})


def parse_tool_params(schema: dict):
    params = {}
    required = schema.get("required", [])
    props = schema.get("properties", {})
    for name, info in props.items():
        params[name] = {
            "type": info.get("type"),
            "desc": info.get("description", ""),
            "required": name in required
        }
    return params

async def list_all_mcp_tools_tcp():
    """TCP 远程MCP客户端查询工具列表"""
    async with client:
        tools = await client.list_tools()
        for t in tools:
            print(f"- {t.name}: {t.description}")
            params_info = parse_tool_params(t.inputSchema)
            print("参数详情：", params_info)

asyncio.run(call_tool())

# asyncio.run(list_all_mcp_tools_tcp())