SmartBizTrip/agent_learn/mcp_base/stdio/client_raw.py

from pathlib import Path
import sys
import asyncio
from langchain_mcp_adapters.tools import load_mcp_tools
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

# 配置mcp服务器脚本路径
server_script = str(Path(__file__).with_name("server_stdio.py").resolve())

# 配置mcp服务器启动参数
server_params = StdioServerParameters(
    command=sys.executable,
    args=[server_script],
)

# 定义mcp客户端
mcp_client = None

# 主要的异步函数run_agent
async def run():
    global mcp_client
    # 启动 MCP server，并通过标准输入输出建立异步连接。
    async with stdio_client(server_params) as (read, write):
        # 使用读写通道创建 MCP 会话。
        async with ClientSession(read, write) as session:
            await session.initialize()
            # 动态创建一个临时类 MCPClientHolder，把 session 放进去。这样就可以在函数外部通过 mcp_client.session 调用 MCP 工具
            mcp_client = type("MCPClientHolder", (), {"session": session})()

            # 从 session 自动获取 MCP server 提供的工具列表。
            tools = await load_mcp_tools(session)
            print(f"tools-->{tools}")

            # 调用 MCP server 的 get_weather 工具
            response = await session.call_tool("get_weather", arguments={})
            print(f"response-->{response}")
    return

# 启动运行
if __name__ == "__main__":
    asyncio.run(run())