mcp协议官方文档

主题

连接本地 MCP 服务器 ​

本指南介绍如何在开发环境中连接和使用本地 MCP 服务器。

概述 ​

本地 MCP 服务器通常通过标准输入/输出 (stdio) 传输方式运行，这是开发和测试阶段最常用的连接方式。

快速开始 ​

1. 安装 MCP 服务器 ​

Python 服务器示例 ​

bash
# 安装 MCP Python SDK
pip install mcp

# 创建简单的服务器
cat > weather_server.py << 'EOF'
import asyncio
from mcp.server import Server
from mcp.server.stdio import stdio_server

app = Server("weather-server")

@app.list_tools()
async def list_tools():
    return [
        {
            "name": "get_weather",
            "description": "Get weather information for a location",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "location": {"type": "string"}
                },
                "required": ["location"]
            }
        }
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "get_weather":
        location = arguments.get("location", "")
        return f"The weather in {location} is sunny and 72°F"
    
    raise ValueError(f"Unknown tool: {name}")

async def main():
    async with stdio_server() as streams:
        await app.run(*streams)

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

TypeScript 服务器示例 ​

bash
# 安装 MCP TypeScript SDK
npm install @modelcontextprotocol/sdk

# 创建服务器文件
cat > weather-server.ts << 'EOF'
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";

const server = new Server(
  {
    name: "weather-server",
    version: "1.0.0",
  },
  {
    capabilities: {
      tools: {},
    },
  }
);

server.setRequestHandler("tools/list", async () => {
  return {
    tools: [
      {
        name: "get_weather",
        description: "Get weather information for a location",
        inputSchema: {
          type: "object",
          properties: {
            location: { type: "string" },
          },
          required: ["location"],
        },
      },
    ],
  };
});

server.setRequestHandler("tools/call", async (request) => {
  const { name, arguments: args } = request.params;
  
  if (name === "get_weather") {
    const location = args.location;
    return {
      content: [
        {
          type: "text",
          text: `The weather in ${location} is sunny and 72°F`,
        },
      ],
    };
  }
  
  throw new Error(`Unknown tool: ${name}`);
});

async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
}

main().catch(console.error);
EOF

# 编译 TypeScript
npx tsc weather-server.ts

2. 配置客户端连接 ​

Claude Desktop 配置 ​

编辑 ~/Library/Application Support/Claude/claude_desktop_config.json:

json
{
  "mcpServers": {
    "weather": {
      "command": "python",
      "args": ["/path/to/weather_server.py"]
    }
  }
}

自定义客户端配置 ​

javascript
const client = new MCPClient({
  name: "example-client",
  version: "1.0.0"
});

await client.connect({
  transport: {
    type: "stdio",
    command: "python",
    args: ["/path/to/weather_server.py"]
  }
});

开发工作流 ​

1. 服务器开发 ​

项目结构 ​

my-mcp-server/
├── src/
│   ├── server.py          # 主服务器文件
│   ├── tools/             # 工具实现
│   │   ├── __init__.py
│   │   ├── weather.py
│   │   └── calculator.py
│   └── resources/         # 资源提供
│       ├── __init__.py
│       └── files.py
├── tests/                 # 测试文件
├── requirements.txt       # 依赖声明
└── README.md             # 文档

实现工具 ​

python
# tools/weather.py
import requests
from typing import Dict, Any

class WeatherTool:
    def __init__(self, api_key: str):
        self.api_key = api_key
    
    async def get_weather(self, location: str) -> Dict[str, Any]:
        """获取指定位置的天气信息"""
        url = f"https://api.openweathermap.org/data/2.5/weather"
        params = {
            "q": location,
            "appid": self.api_key,
            "units": "metric"
        }
        
        response = requests.get(url, params=params)
        response.raise_for_status()
        
        data = response.json()
        return {
            "location": data["name"],
            "temperature": data["main"]["temp"],
            "description": data["weather"][0]["description"],
            "humidity": data["main"]["humidity"]
        }

实现资源 ​

python
# resources/files.py
import os
from pathlib import Path
from typing import List, Dict, Any

class FileResource:
    def __init__(self, base_path: str):
        self.base_path = Path(base_path)
    
    async def list_files(self) -> List[Dict[str, Any]]:
        """列出可用的文件资源"""
        resources = []
        for file_path in self.base_path.rglob("*.txt"):
            resources.append({
                "uri": f"file://{file_path}",
                "name": file_path.name,
                "description": f"Text file: {file_path.name}",
                "mimeType": "text/plain"
            })
        return resources
    
    async def read_file(self, uri: str) -> str:
        """读取文件内容"""
        file_path = Path(uri.replace("file://", ""))
        if not file_path.is_relative_to(self.base_path):
            raise ValueError("Access denied: file outside allowed directory")
        
        return file_path.read_text(encoding="utf-8")

2. 测试和调试 ​

使用 MCP Inspector ​

bash
# 安装 MCP Inspector
npm install -g @modelcontextprotocol/inspector

# 启动 Inspector 连接到本地服务器
mcp-inspector python /path/to/weather_server.py

单元测试 ​

python
# tests/test_weather.py
import pytest
from unittest.mock import patch, Mock
from src.tools.weather import WeatherTool

@pytest.fixture
def weather_tool():
    return WeatherTool(api_key="test_key")

@patch('requests.get')
async def test_get_weather(mock_get, weather_tool):
    # 模拟 API 响应
    mock_response = Mock()
    mock_response.json.return_value = {
        "name": "San Francisco",
        "main": {"temp": 20, "humidity": 65},
        "weather": [{"description": "clear sky"}]
    }
    mock_get.return_value = mock_response
    
    # 测试工具调用
    result = await weather_tool.get_weather("San Francisco")
    
    assert result["location"] == "San Francisco"
    assert result["temperature"] == 20
    assert result["description"] == "clear sky"

集成测试 ​

python
# tests/test_integration.py
import asyncio
import json
from src.server import create_server

async def test_server_integration():
    server = create_server()
    
    # 模拟初始化请求
    init_request = {
        "jsonrpc": "2.0",
        "method": "initialize",
        "params": {
            "protocolVersion": "2025-06-18",
            "capabilities": {},
            "clientInfo": {"name": "test-client", "version": "1.0.0"}
        },
        "id": 1
    }
    
    response = await server.handle_request(init_request)
    assert response["result"]["protocolVersion"] == "2025-06-18"

调试技巧 ​

1. 日志记录 ​

python
import logging

# 配置日志
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('/tmp/mcp_server.log'),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)

# 在代码中添加日志
@app.call_tool()
async def call_tool(name: str, arguments: dict):
    logger.info(f"Tool called: {name} with args: {arguments}")
    try:
        result = await execute_tool(name, arguments)
        logger.info(f"Tool result: {result}")
        return result
    except Exception as e:
        logger.error(f"Tool execution failed: {e}")
        raise

2. 错误处理 ​

python
from mcp.server.exceptions import McpError

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    try:
        if name == "get_weather":
            location = arguments.get("location")
            if not location:
                raise McpError(
                    code=-32602,
                    message="Invalid params",
                    data={"param": "location", "error": "Location is required"}
                )
            
            return await get_weather_data(location)
        else:
            raise McpError(
                code=-32601,
                message="Method not found",
                data={"method": name}
            )
    except Exception as e:
        logger.exception("Unexpected error in tool execution")
        raise McpError(
            code=-32603,
            message="Internal error",
            data={"error": str(e)}
        )

3. 性能监控 ​

python
import time
from functools import wraps

def monitor_performance(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        start_time = time.time()
        try:
            result = await func(*args, **kwargs)
            duration = time.time() - start_time
            logger.info(f"{func.__name__} completed in {duration:.3f}s")
            return result
        except Exception as e:
            duration = time.time() - start_time
            logger.error(f"{func.__name__} failed after {duration:.3f}s: {e}")
            raise
    return wrapper

@monitor_performance
async def get_weather_data(location: str):
    # 实现获取天气数据的逻辑
    pass

常见问题 ​

1. 连接失败 ​

问题: 客户端无法连接到服务器 解决方案:

  • 检查服务器脚本路径是否正确
  • 确认 Python/Node.js 环境配置
  • 查看服务器日志文件

2. 工具调用失败 ​

问题: 工具调用返回错误 解决方案:

  • 验证工具参数格式
  • 检查工具实现逻辑
  • 添加详细的错误日志

3. 性能问题 ​

问题: 服务器响应缓慢 解决方案:

  • 使用异步操作避免阻塞
  • 实现结果缓存机制
  • 优化数据库查询

最佳实践 ​

1. 代码组织 ​

  • 使用模块化设计分离关注点
  • 实现清晰的错误处理策略
  • 添加全面的测试覆盖

2. 安全考虑 ​

  • 验证所有输入参数
  • 限制文件系统访问范围
  • 使用环境变量管理敏感信息

3. 性能优化 ​

  • 实现异步操作
  • 使用连接池管理外部资源
  • 添加适当的缓存机制

相关资源 ​

🚀 探索模型上下文协议的无限可能 | 连接 AI 与世界的桥梁 | 让智能更智能

Copyright © 2025 mcp协议官方文档 ｜ sitemap