基于Python从零构建一个MCP服务器

目录

• 一、MCP协议：AI与工具的"USB-C接口"
• 二、环境搭建：3分钟启动开发环境
• 三、工具开发：让AI可调用的"魔法函数"
• 四、服务器构建：3种通信模式详解
• 五、AI集成：让Claude调用你的工具
• 六、安全与性能优化
• 七、完整项目示例：股票数据查询系统
• 八、未来展望：MCP生态演进方向

想象你正在用AI助手处理工作：需要查询天气时，AI突然弹出"我需要调用天气API"的提示；处理Excel数据时，它又卡在"如何读取CSV文件"的步骤。传统AI的局限性在于，它像一台没有外设的电脑——能思考却无法操作现实世界。

MCP（Model Context Protocol）协议的出现解决了这个痛点。它定义了一套标准化通信机制，让AI代理（如Claude、Cursor）能动态发现并调用服务器上的工具函数。就像给电脑插上USB-C线，AI瞬间获得了访问数据库、调用API、操作文件系统的能力。

核心价值：

• 解耦设计：工具函数与AI模型分离，修改工具不影响模型训练
• 安全沙箱：通过服务器中转调用，避免直接暴露API密钥
• 统一入口：用标准化协议整合分散的工具接口

创建隔离环境

# 使用uv工具快速初始化项目（推荐）
uv init mcp_weather_server && cd mcp_weather_server
uv venv && source .venv/bin/activate  # Linux/Mac
# 或 .venvScriptsactivate (Windows)

安装核心依赖

# 基础版（适合个人开发）
uv add mcp>=1.9.1 requests python-dotenv
 
# 企业版（需处理高并发）
uv add fastmcp httpx uvicorn[standard]

验证安装

# 创建test_install.py
from mcp.server.fastmcp import FastMCP
print("MCP SDK安装成功" if 'FastMCP' in globals() else "安装失败")

运行 python test_install.py 应输出成功信息。

案例1：实时天气查询工具

# weather_tool.py
import requests
from mcp.server.fastmcp import tool
from dotenv import load_dotenv
import os
 
load_dotenv()  # 加载环境变量
 
@tool()
def get_weather(city: str) -> dict:
    """获取城市实时天气（需配置API_KEY）
    
    Args:
        city: 城市名称（如"北京"）
    
    Returns:
        包含温度、湿度、天气状况的字典
    """
    api_key = os.getenv('WEATHER_API_KEY')
    url = f"https://api.openweathermap.org/data/2.5/weather?q={city}&appid={api_key}&units=metric"
    
    try:
        response = requests.get(url, timeout=5)
        data = response.json()
        return {
            "temperature": data["main"]["temp"],
            "humidity": data["main"]["humidity"],
            "condition": data["weather"][0]["description"]
        }
    except Exception as e:
        return {"error": f"天气查询失败: {str(e)}"}

关键设计原则：

• 类型注解：用 city: str 明确参数类型
• 错误处理：捕获网络请求异常
• 文档字符串：详细说明参数和返回值
• 环境变量：敏感信息（如API密钥）通过 .env 文件管理

案例2：数据库查询工具

# db_tool.py
import sqlite3
from mcp.server.fastmcp import tool
from mcp.types import TableContent
 
@tool()
def query_sales_data(limit: int = 10) -> TableContent:
    """查询销售数据表（示例用）
    
    Args:
        limit: 返回记录数上限
    
    Returns:
        表格数据（包含列名和行数据）
    """
    conn = sqlite3.connect("sales.db")
    cursor = conn.cursor()
    cursor.execute(f"SELECT * FROM orders LIMIT {limit}")
    
    columns = [desc[0] for desc in cursor.description]
    rows = cursor.fetchall()
    
    return TableContent(
        type="table",
        columns=columns,
        rows=rows
    )

企业级优化：

• 使用参数化查询防止SQL注入
• 添加连接池管理数据库连接
• 通过Nacos动态配置数据库地址

模式1：本地开发模式（stdio）

# server_stdio.py
from mcp.server.fastmcp import FastMCP
from weather_tool import get_weather
from db_tool import query_sales_data
 
mcp = FastMCP("Local Tools Server")
mcp.add_tool(get_weather)
mcp.add_tool(query_sales_data)
 
if __name__ == "__main__":
    mcp.run(mode="stdio")  # 通过标准输入输出通信

适用场景：

• 与Claude Desktop/Cursor本地集成
• 快速验证工具功能
• 开发阶段调试

模式2：云端部署模式（SSE）

# server_sse.py
from fastmcp import FastMCP
from weather_tool import get_weather
import uvicorn
 
mcp = FastMCP("Cloud Tools Server")
mcp.add_tool(get_weather)
 
if __name__ == "__main__":
    # 通过HTTP事件流通信
    uvicorn.run(
        mcp.as_app(),  # 转换为FastAPI应用
        host="0.0.0.0",
        port=8000,
        workers=4  # 多进程处理并发请求
    )

关键配置：

• 添加JWT认证中间件
• 设置CORS允许前端调用
• 配置Nginx反向代理

模式3：自动化迁移模式（OpenAPI转换）

# converter.py
import json
from jsonschema import validate
 
def convert_to_mcp_tool(api_spec):
    """将OpenAPI接口转换为MCP工具
    
    Args:
        api_spec: OpenAPI规范字典
    
    Returns:
        MCP工具代码模板字符串
    """
    tool_template = f"""
@tool()
def {api_spec["operationId"]}(**kwargs):
    """
    {api_spec["summary"]}
    
    Parameters:
        {json.dumps(api_spec["parameters"], indent=4)}
    """
    # 这里添加实际调用逻辑
    return盛{{}}
    """
    return tool_template
 
# 示例：转换天气API
weather_api = {
    "operationId": "get_weather",
    "summary": "获取城市天气信息",
    "parameters": [
        {"name": "city", "in": "query", "required": True, "type": "string"}
    ]
}
print(convert_to_mcp_tool(weather_api))

企业应用：

• 批量迁移现有RESTful API
• 生成标准化工具文档
• 与Higress网关集成实现协议转换

1.配置Claude Desktop

编辑配置文件 claude_desktop_config.json：

{
  "mcpServers": {
    "weather_service": {
      "command": "python",
      "args": ["server_stdio.py", "--mode", "stdio"]
    }
  }
}

2.自然语言调用示例

当用户在Claude中输入：

"查询北京今天的天气"

AI内部处理流程：

发现可用的 get_weather 工具

将自然语言转换为工具调用：

{
  "toolName": "get_weather",
  "arguments": {"city": "北京"}
}

执行工具并返回结果：

{
  "temperature": 28,
  "humidity": 65,
  "condition": "多云"
}

将结果生成自然语言回复：

"北京今天气温28℃，湿度65%，天气多云"

3.调试技巧

使用MCP Inspector可视化调试：

npx @modelcontextprotocol/inspector python server_stdio.py

浏览器打开 http://localhost:3000 可实时查看：

• 工具注册情况
• 请求/响应数据流
• 调用耗时统计

安全防护三板斧

敏感信息隔离：

# 使用Vault管理密钥
from hvac import Client
 
def get_secret(path: str) -> str:
    client = Client(url="http://vault-server:8200")
    return client.secrets.kv.v2.read_secret_version(path=path)["data"]["data"]["key"]

参数校验：

from pydantic import BaseModel, conint
 
class QueryParams(BaseModel):
    limit: conint(ge=1, le=100)  # 限制1-100的整数
 
@tool()
def safe_query(params: QueryParams) -> dict:
    validated = params.dict()  # 自动校验参数
    # 执行查询逻辑...

执行超时控制：

from functools import partial
import asyncio
 
async def timed_fetch(url: str, timeout: float = 5.0):
    try:
        return await asyncio.wait_for(requests.get(url), timeout=timeout)
    except asyncio.TimeoutError:
        return {"error": "请求超时"}

性能优化方案

缓存策略：

from functools import lru_cache
 
@lru_cache(maxsize=100)
@tool()
def cached_weather(city: str) -> dict:
    # 首次调用执行实际查询
    # 后续调用直接返回缓存结果
    return _real_weather_query(city)

异步处理：

@tool()
async def async_data_processing(file_url: str):
    async with httpx.AsyncClient() as client:
        file_content = await client.get(file_url)
        # 并行处理数据...

水平扩展：

# 使用Gunicorn部署FastMCP应用
gunicorn server_sse:app -w 8 -k uvicorn.workers.UvicornWorker

项目结构

stock_mcp/
├── .env                # API密钥配置
├── tools/
│   ├── __init__.py
│   ├── stock_api.py    # 股票查询工具
│   └── data_analysis.py # 数据分析工具
├── server.py           # 服务器入口
└── requirements.txt    # 依赖列表

核心代码实现

# tools/stock_api.py
import httpx
from mcp.server.fastmcp import tool
from dotenv import load_dotenv
import os
 
load_dotenv()
 
@tool()
def get_stock_price(symbol: str) -> dict:
    """获取股票实时价格（Alpha Vantage API）
    
    Args:
        symbol: 股票代码（如"AAPL"）
    
    Returns:
        包含价格、涨跌幅的字典
    """
    api_key = os.getenv('ALPHA_VANTAGE_KEY')
    url = f"https://www.alphavantage.co/query?function=GLOBAL_QUOTE&symbol={symbol}&apikey={api_key}"
    
    response = httpx.get(url)
    data = response.json()
    
    if "Global Quote" not in data:
        return {"error": "未找到股票数据"}
    
    quote = data["Global Quote"]
    return {
        "symbol": symbol,
        "price": float(quote["05. price"]),
        "change": float(quote["09. change"]),
        "change_percent": float(quote["10. change percent"].strip('%'))
    }
 
# server.py
from fastmcp import FastMCP
from tools.stock_api import get_stock_price
from tools.data_analysis import calculate_stats
 
mcp = FastMCP("Stock Analysis Server")
mcp.add_tool(get_stock_price)
mcp.add_tool(calculate_stats)
 
if __name__ == "__main__":
    mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)

部署与测试

# 安装依赖
uv install -r requirements.txt
 
# 启动服务器
python server.py
 
# 测试工具调用
curl -X POST http://localhost:8000/mcp/invoke 
  -H "Content-Type: application/json" 
  -d '{"toolName": "get_stock_price", "arguments": {"symbol": "AAPL"}}'

• 工具市场：类似PyPI的MCP工具仓库，实现"一键安装"工具
• 协议扩展：支持gRPC、WebSocket等更多通信协议
• 智能路由：根据请求内容自动选择最优工具
• 边缘计算：在IoT设备上部署轻量级MCP服务器

MCP协议正在重塑AI开发范式——它让大模型从"封闭的大脑"进化为"可连接万物的神经系统"。无论是个人开发者快速扩展AI能力，还是企业整合遗留系统，MCP都提供了标准化解决方案。当工具调用变得像呼吸一样自然，AI才能真正成为生产力的延伸。

以上就是基于Python从零构建一个MCP服务器的详细内容，更多关于Python MCP服务器的资料请关注风君子博客其它相关文章！

您可能感兴趣的文章:

• 使用Python构建MCP服务器的详细配置步骤
• Python使用FastMCP实现Word文档与JSON数据互转
• Python FastMCP构建MCP服务端与客户端的详细步骤