python如何连接免费的mcp服务？

我将为你提供一个使用 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(sel...

点击查看剩余70%