使用 Python SDK 开发 MCP 客户端

MCP 客户端开发——使用 Python SDK

简介与设置

本教程将指导您使用 Python SDK 开发 MCP (Model Context Protocol，模型上下文协议) 客户端。MCP 客户端连接到 MCP 服务器，以利用其提供的资源 (Resources)、提示 (Prompts) 和工具 (Tools)，这些是构建上下文感知型 AI 应用的基础。

Python SDK 提供了一个 Pythonic 接口，利用 asyncio 实现非阻塞操作，非常适合用于构建需要与 MCP 服务器交互的后端服务、脚本或任何 Python 应用程序。

先决条件：

• Python (推荐 3.8 或更高版本)。
• pip (Python 的包安装器)。

安装：

从 PyPI 安装 MCP Python SDK 包：

pip install modelcontext-mcp

连接到 MCP 服务器

MCPClient 类用于管理连接。最佳实践是使用 async with 语句，它能确保连接的正确建立和关闭。

import asyncio
from modelcontext_mcp import MCPClient, MCPError

async def connect_and_operate():
    server_uri = 'ws://localhost:8080/mcp'  # 替换为您的 MCP 服务器的 WebSocket URI
    try:
        async with MCPClient(
            uri=server_uri,
            client_id='my-python-client-01',
            client_description='An example MCP client using the Python SDK'
            # 您还可以在此处声明客户端能力
        ) as client:
            print(f'Successfully connected to MCP server at {server_uri}!')
            print(f'Server capabilities: {client.server_capabilities}')

            # 客户端现已连接并准备好进行操作。
            # 我们将在下一节中添加操作。
            pass # 操作的占位符

    except MCPError as e:
        print(f'MCP Client Error (Code: {e.code}): {e.message}. Data: {e.data}')
    except ConnectionRefusedError:
        print(f'Connection refused. Is the MCP server running at {server_uri}?')
    except Exception as e:
        print(f'An unexpected error occurred: {e}')
    finally:
        print('Client operations finished or an error occurred.')

# 要运行此示例（以及后续示例）：
# if __name__ == "__main__":
#     asyncio.run(connect_and_operate())

async with 块会在进入时自动处理 client.connect()，并在退出时自动处理 client.close()。

与服务器能力交互

建立活动的客户端连接后，您可以使用客户端的 request_* 和 stream_* 方法与服务器的资源 (Resources)、提示 (Prompts) 和工具 (Tools) 进行交互。

请求资源 (Resources)： 要从资源中获取数据，请使用 client.request_resource(resource_id, params) 获取单个响应，或迭代 client.stream_resource(resource_id, params) 以获取流式数据。

# Inside an async function, assuming 'client' is an active MCPClient instance

async def get_server_file(client: MCPClient):
    try:
        # 假设服务器有一个名为 'file_content' 的资源
        # 它接受 {'path': 'file_path.txt'} 并返回 {'content': '...'}
        response = await client.request_resource(
            resource_id='file_content',
            params={'path': '/data/info.txt'}
        )
        print(f"Resource (file_content) response: {response.payload.get('content')}")
    except MCPError as e:
        print(f'Error requesting resource: {e.message}')

async def stream_server_logs(client: MCPClient):
    try:
        # 假设服务器有一个名为 'live_logs' 的资源，它流式传输 {'line': 'log line'}
        print('Streaming server logs...')
        async for update in client.stream_resource(
            resource_id='live_logs',
            params={'level': 'INFO'}
        ):
            print(f"Log: {update.payload.get('line')}")
            if "END_STREAM" in update.payload.get('line', ''):
                 print("End of stream marker found.")
                 break
        print('Log streaming finished.')
    except MCPError as e:
        print(f'Error streaming resource: {e.message}')

利用提示 (Prompts)： 要向提示 (例如，用于 LLM 处理) 发送数据，请使用 client.request_prompt(prompt_id, params) 获取单个输出，或使用 client.stream_prompt(prompt_id, params) 获取流式输出，例如逐令牌 (token) 生成。

# Inside an async function, assuming 'client' is an active MCPClient instance

async def get_text_completion(client: MCPClient, user_input: str):
    try:
        # 假设 'text_generator' 提示接受 {'input_text': '...'}
        # 并返回 {'completion': '...'}
        response = await client.request_prompt(
            prompt_id='text_generator',
            params={'input_text': user_input, 'max_tokens': 50}
        )
        print(f"Prompt (text_generator) completion: {response.payload.get('completion')}")
    except MCPError as e:
        print(f'Error requesting prompt: {e.message}')

async def stream_generated_text(client: MCPClient, topic: str):
    try:
        # 假设 'creative_writer' 流式传输 {'token': '...'}
        print(f'Streaming creative writing on "{topic}"...')
        full_text = []
        async for update in client.stream_prompt(
            prompt_id='creative_writer',
            params={'topic': topic}
        ):
            token = update.payload.get('token', '')
            print(token, end='', flush=True)
            full_text.append(token)
        print('\nCreative writing stream finished.')
        # print(f"Full text: {''.join(full_text)}")
    except MCPError as e:
        print(f'Error streaming prompt: {e.message}')

调用工具 (Tools)： 要执行服务器端函数，请使用 client.request_tool(tool_id, params)。

# Inside an async function, assuming 'client' is an active MCPClient instance

async def execute_remote_calculation(client: MCPClient, num1: int, num2: int):
    try:
        # 假设 'math_tool' 接受 {'a': N, 'b': N, 'op': 'name'}
        # 并返回 {'result': M}
        response = await client.request_tool(
            tool_id='math_tool',
            params={'a': num1, 'b': num2, 'op': 'multiply'}
        )
        print(f"Tool (math_tool) result: {response.payload.get('result')}")
    except MCPError as e:
        print(f'Error invoking tool: {e.message}')

错误处理

正确的错误处理至关重要。捕获 MCPError 以处理协议特定问题（例如无效参数或服务器端执行错误），并捕获标准 Python 异常（例如 ConnectionRefusedError）以处理网络或其他问题。

# General structure for error handling in an operation
# client: MCPClient
try:
    # response = await client.request_resource(...)
    pass # 您的 MCP 操作在此处
except MCPError as e:
    # 来自服务器的特定 MCP 错误或协议违规
    print(f'MCP Error! Code: {e.code}, Message: {e.message}, Details: {e.data}')
except ConnectionRefusedError:
    print('Connection was refused by the server. Is it running and accessible?')
except asyncio.TimeoutError:
    print('The operation timed out.')
except Exception as e:
    # 其他意外错误
    print(f'An unexpected Python error occurred: {type(e).__name__} - {e}')

完整客户端示例

本示例结合了以下概念：连接到服务器、请求资源、利用提示、调用工具以及处理潜在错误。它假设一个 MCP 服务器正在 ws://localhost:8080/mcp 运行并公开了指定的能力。

import asyncio
from modelcontext_mcp import MCPClient, MCPError

async def run_full_python_client():
    server_uri = 'ws://localhost:8080/mcp'
    try:
        async with MCPClient(
            uri=server_uri,
            client_id='py-full-client-example',
            client_description='Full Python MCP Client Example'
        ) as client:
            print(f'Connected to {server_uri}. Server Info: {client.server_capabilities}')

            # 1. 请求 'system_status' 资源
            # 假设 payload: {'status': '...', 'load': 0.0}
            print("\nRequesting 'system_status' resource...")
            status_payload = await client.request_resource(resource_id='system_status', params={})
            print(f"System Status: {status_payload.payload.get('status')}, Load: {status_payload.payload.get('load')}")

            # 2. 利用 'translate' 提示
            # 假设 params: {'text': '...', 'target_lang': '...'}, payload: {'translation': '...'}
            text_to_translate = "Hello, MCP world!"
            print(f"\nRequesting 'translate' prompt for: '{text_to_translate}' to Spanish...")
            translate_payload = await client.request_prompt(
                prompt_id='translate',
                params={'text': text_to_translate, 'target_lang': 'es'}
            )
            print(f"Translation: {translate_payload.payload.get('translation')}")

            # 3. 调用 'database_query' 工具
            # 假设 params: {'query': 'SELECT ...'}, payload: {'rows': [...]}
            print("\nInvoking 'database_query' tool...")
            query_result = await client.request_tool(
                tool_id='database_query',
                params={'query': 'SELECT COUNT(*) FROM users;'}
            )
            print(f"Query Result (User Count): {query_result.payload.get('rows')}")

    except MCPError as e:
        print(f'MCP Communication Error! Code: {e.code}, Message: {e.message}, Data: {e.data}')
    except ConnectionRefusedError:
        print(f'Connection refused at {server_uri}. Please ensure the MCP server is running.')
    except Exception as e:
        print(f'An unexpected error occurred: {type(e).__name__} - {e}')
    finally:
        print('\nPython client operations concluded.')

if __name__ == "__main__":
    asyncio.run(run_full_python_client())

Python MCP 客户端的最佳实践

• 上下文管理： 始终使用 async with MCPClient(...) as client: 来确保连接的正确建立和关闭，即使发生错误也不例外。
• 全面的错误处理： 专门捕获 MCPError 以处理协议层面的问题，并准备好处理常见的网络异常，例如 ConnectionRefusedError 或 asyncio.TimeoutError。
• 异步代码： 编写所有与 MCP 服务器交互的客户端代码时，请使用 async 和 await，以利用 Python 的 asyncio 库实现高效的非阻塞 I/O。
• Payload 验证： 尽管 Python 是动态类型语言，但仍应考虑验证 response.payload 字典中预期键的结构或存在性，尤其是在与不熟悉的服务器交互时。
• 流式传输以提高效率： 在处理潜在的大型数据集或实时数据流时，使用 client.stream_resource 和 client.stream_prompt 可以避免高内存使用并提高响应速度。
• 理解服务器能力： 连接后检查 client.server_capabilities，以编程方式了解所连接服务器提供的功能。这可以使您的客户端更具适应性。