Python 连接 MCP Server 实战指南

方案 B：使用 Streamable HTTP 连接

这是更底层的连接方式，通常需要自行管理 httpx.AsyncClient 的生命周期。

import httpx
from mcp.client.streamable_http import streamable_http_client
from mcp.client.session import ClientSession

async def connect_via_http(url: str, headers: dict = None):
    # 必须显式创建 httpx client，以便复用连接池和配置超时
    async with httpx.AsyncClient(headers=headers, timeout=60.0) as http_client:
        # streamable_http_client 同样返回读写流
        async with streamable_http_client(url=url, http_client=http_client) as (read, write, _):
            async with ClientSession(read, write) as session:
                yield session

3. 核心交互流程

一旦 ClientSession 建立，后续的操作就是标准的 MCP 协议交互。

3.1 握手与初始化 (Initialize)

连接建立后的第一件事必须是握手。Server 会在此时返回它的能力（Capabilities）和元数据（Server Info）。

# 初始化会话
# 这一步会协商协议版本和能力
init_result = await session.initialize()
print(f"Connected to Server: {init_result.serverInfo.name} v{init_result.serverInfo.version}")
# 技巧：这里往往藏着 Server 的 System Prompt
if hasattr(init_result, 'instructions'):
    print(f"Server Instructions: {init_result.instructions}")

3.2 工具发现 (List Tools)

这是 Agent'获取技能'的关键步骤。

# 获取工具列表
list_tools_result = await session.list_tools()
tools = list_tools_result.tools
for tool in tools:
    print(f"Found Tool: {tool.name}")
    print(f"Description: {tool.description}")
    print(f"Schema: {tool.inputSchema}")
# JSON Schema 格式的参数定义

工程提示：在实际应用中，你需要将这里的 inputSchema 转换为你的 LLM 框架（如 LangChain 或 OpenAI SDK）所需的格式。对于 LangChain，可以使用 pydantic.create_model 动态生成参数模型。

3.3 工具调用 (Call Tool)

当 LLM 决定调用工具时，Client 负责转发请求。

from mcp.types import CallToolResult

tool_name = "get_repo_list"
arguments = {"owner": "nvd11", "limit": 5}

try:
    # 发送调用请求
    result: CallToolResult = await session.call_tool(tool_name, arguments=arguments)
    # 解析结果
    # MCP 支持 Text 和 Image 两种内容类型
    output_text = ""
    for content in result.content:
        if content.type == "text":
            output_text += content.text
        elif content.type == "image":
            print(f"[Image Content: {content.mimeType}]")
    print(f"Tool Output: {output_text}")
except Exception as e:
    print(f"Tool execution failed: {e}")

4. 常见问题 (FAQ)

Q: 我可以用 aiohttp 代替 httpx 吗？

A: 不可以直接替换。
mcp.client.streamable_http.streamable_http_client 显式依赖于 httpx.AsyncClient 的接口。由于 mcp SDK 本身就将 httpx 作为核心依赖，建议遵循官方实践使用 httpx，以避免不必要的适配工作。

5. 总结

对于大多数 Python 后端服务（如 Agent 平台），Streamable HTTP 提供了更好的控制力（如自定义 Timeout、Proxy 配置），是更为推荐的选择。