回答

我将为你提供一个使用 Python 调用 Claude API 并集成 MCP (Model Context Protocol) 服务的示例代码。

这个例子展示如何创建一个简单的 MCP 客户端，连接到一个 MCP 服务端，并通过 Claude API 调用其功能。

import asyncio
import logging
from typing import Optional, Dict, Any
from anthropic import AsyncAnthropic
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import os

# 设置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class MCPClaudeClient:
    def __init__(self, api_key: str, server_script_path: str):
        """初始化 MCPClaude 客户端
        
        Args:
            api_key: Anthropic API 密钥
            server_script_path: MCP 服务端脚本路径
        """
        self.claude = AsyncAnthropic(api_key=api_key)
        self.server_script_path = server_script_path
        self.session: Optional[ClientSession] = None

    async def connect_to_server(self, max_retries: int = 3) -> None:
        """连接到 MCP 服务端
        
        Args:
            max_retries: 最大重试次数
            
        Raises:
            ConnectionError: 如果连接失败
        """
        # 配置服务端参数
        server_params = StdioServerParameters(
            command="python",
            args=[self.server_script_path],
            env=None
        )

        # 重试机制
        last_exception = None
        for attempt in range(max_retries):
            try:
                # 创建并连接 MCP 会话
                read, write = await stdio_client(server_params)
                self.session = ClientSession(read, write)
                await self.session.initialize()

                # 列出可用工具
                tools_response = await self.session.list_tools()
                logger.info(f"可用工具: {[tool.name for tool in tools_response.tools]}")
                return
                
            except Exception as e:
                last_exception = e
                logger.warning(f"连接尝试 {attempt + 1}/{max_retries} 失败: {str(e)}")
                if attempt < max_retries - 1:
                    await asyncio.sleep(1 * (attempt + 1))  # 指数退避
        
        raise ConnectionError(f"无法连接到 MCP 服务端: {str(last_exception)}")

    async def process_query(self, query: str) -> str:
        """使用 Claude 和 MCP 处理查询
        
        Args:
            query: 用户查询文本
            
        Returns:
            处理后的响应文本
            
        Raises:
            RuntimeError: 如果会话未初始化或处理失败
        """
        if not self.session:
            raise RuntimeError("会话未初始化，请先调用 connect_to_server()")
            
        try:
            # 获取可用工具
            tools_response = await self.session.list_tools()
            available_tools = [{
                "name": tool.name,
                "description": tool.description,
                "input_schema": tool.inputSchema
            } for tool in tools_response.tools]

            # 构造初始消息
            messages = [{"role": "user", "content": query}]

            # 调用 Claude API
            response = await self.claude.messages.create(
                model="claude-3-sonnet-20240229",
                max_tokens=1000,
                messages=messages,
                tools=available_tools
            )

            # 处理响应和工具调用
            final_text = []
            for content in response.content:
                if content.type == 'text':
                    final_text.append(content.text)
                elif content.type == 'tool_use':
                    # 执行 MCP 工具调用
                    tool_name = content.name
                    tool_args = content.input
                    try:
                        result = await self.session.call_tool(tool_name, tool_args)
                        if hasattr(result, 'content'):
                            final_text.append(f"[工具调用 {tool_name}: {result.content}]")
                        else:
                            final_text.append(f"[工具调用 {tool_name} 完成，但未返回内容]")
                    except Exception as e:
                        final_text.append(f"[工具调用 {tool_name} 失败: {str(e)}]")
                        logger.error(f"工具调用 {tool_name} 失败: {str(e)}")

            return "\n".join(final_text)
            
        except Exception as e:
            logger.error(f"处理查询失败: {str(e)}")
            raise RuntimeError(f"处理查询时出错: {str(e)}")

    async def cleanup(self) -> None:
        """清理资源"""
        try:
            if self.session:
                await self.session.close()
                self.session = None
        except Exception as e:
            logger.error(f"清理资源时出错: {str(e)}")

async def main():
    try:
        # 配置你的 Anthropic API Key 和 MCP 服务端脚本路径
        API_KEY = os.getenv("ANTHROPIC_API_KEY", "your-anthropic-api-key")  # 优先从环境变量获取
        SERVER_PATH = os.getenv("MCP_SERVER_PATH", "path/to/your/mcp_server.py")  # 优先从环境变量获取

        if API_KEY == "your-anthropic-api-key" or SERVER_PATH == "path/to/your/mcp_server.py":
            raise ValueError("请配置正确的 API_KEY 和 SERVER_PATH")

        # 创建客户端实例
        client = MCPClaudeClient(API_KEY, SERVER_PATH)

        try:
            # 连接到 MCP 服务端
            await client.connect_to_server()

            # 示例查询
            query = "请帮我获取当前天气信息"
            response = await client.process_query(query)
            print("响应:", response)

        finally:
            await client.cleanup()
            
    except Exception as e:
        logger.error(f"程序运行失败: {str(e)}")
        raise

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

前置条件

安装依赖：

pip install anthropic mcp

准备工作：

需要一个有效的 Anthropic API Key（可从 Anthropic 官网获取）需要一个运行的 MCP 服务端脚本（mcp_server.py），可以是任何实现了 MCP 协议的服务端，例如天气服务、文件系统服务等代码说明

MCPClaudeClient 类：

初始化时需要提供 Claude API Key 和 MCP 服务端脚本路径connect_to_server: 建立与 MCP 服务端的连接process_query: 处理用户查询，通过 Claude API 调用并结合 MCP 工具cleanup: 清理资源

工具调用流程：

获取 MCP 服务端提供的可用工具将工具信息传递给 Claude API当 Claude 决定使用工具时，通过 MCP 会话执行工具调用

示例 MCP 服务端以下是一个简单的 MCP 服务端示例（保存为 mcp_server.py）：

from mcp import Server

mcp = Server("weather-server")

@mcp.tool()
async def get_weather(city: str) -> str:
    """获取指定城市的天气"""
    return f"{city} 的当前天气：晴天，25°C"

if __name__ == "__main__":
    mcp.run(transport='stdio')

使用方法将 API Key 替换为你的实际 Key将 SERVER_PATH 替换为你的 MCP 服务端脚本路径运行代码：

python your_script.py

输出示例

可用工具: ['get_weather']
响应: [工具调用 get_weather: 上海 的当前天气：晴天，25°C]

注意事项确保 MCP 服务端脚本已正确实现并可运行Claude API 调用需要稳定的网络连接根据实际需求调整 model 参数和 max_tokens 值

这个示例展示了如何将 Claude API 与 MCP 服务结合使用，你可以根据需要扩展功能，例如添加更复杂的工具或处理多轮对话。