在 Python 中调用 MCP 客户端，通常涉及使用 SDK 或直接与 MCP 服务进行通信。以下是一个基于 SDK 的示例，展示如何实现 MCP 客户端的调用。 ### 实现 MCP 客户端调用 以下代码展示了如何创建一个支持流式响应的 MCP 客户端类 `MCPStreamingClient`，并实现流式生成内容的功能： ```python import json import requests from typing import Any class MCPTaskType: TEXT_GENERATION = "text_generation" class MCPClient: def __init__(self, api_endpoint: str, api_key: str): self.api_endpoint = api_endpoint self.api_key = api_key def create_mcp_request(self, prompt: str, task_type: MCPTaskType, **kwargs) -> dict: return { "prompt": prompt, "task_type": task_type, "model_control": kwargs.get("model_control", {}), } class MCPStreamingClient(MCPClient): """支持流式响应的MCP客户端""" def stream_generate(self, prompt: str, task_type: MCPTaskType = MCPTaskType.TEXT_GENERATION, **kwargs) -> Any: """流式生成内容""" request = self.create_mcp_request( prompt=prompt, task_type=task_type, **kwargs ) # 添加流式参数 request["model_control"]["streaming"] = True headers = { "Content-Type": "application/json", "Authorization": f"Bearer {self.api_key}" } with requests.post( self.api_endpoint, headers=headers, data=json.dumps(request), stream=True ) as response: response.raise_for_status() for line in response.iter_lines(): if line: yield json.loads(line.decode('utf-8')) ``` ### 使用示例 假设你已经有一个 MCP 服务的 API 端点和 API 密钥，可以按照如下方式使用上面定义的 `MCPStreamingClient` 类来调用 MCP 客户端： ```python if __name__ == "__main__": api_endpoint = "https://your-mcp-api-endpoint.com" api_key = "your-api-key" client = MCPStreamingClient(api_endpoint=api_endpoint, api_key=api_key) prompt = "Hello, how can I assist you today?" for response_line in client.stream_generate(prompt): print(response_line) ``` 上述代码将向 MCP 服务发送请求，并以流式方式接收响应。 ---