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

一、mcp协议：ai与工具的"usb-c接口"

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

mcp（model context protocol）协议的出现解决了这个痛点。它定义了一套标准化通信机制，让ai代理（如claude、cursor）能动态发现并调用服务器上的工具函数。就像给电脑插上usb-c线，ai瞬间获得了访问数据库、调用api、操作文件系统的能力。

核心价值：

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

二、环境搭建：3分钟启动开发环境

创建隔离环境

# 使用uv工具快速初始化项目（推荐）
uv init mcp_weather_server && cd mcp_weather_server
uv venv && source .venv/bin/activate  # linux/mac
# 或 .venv\scripts\activate (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 应输出成功信息。

三、工具开发：让ai可调用的"魔法函数"

案例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动态配置数据库地址

四、服务器构建：3种通信模式详解

模式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网关集成实现协议转换

五、ai集成：让claude调用你的工具

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"}}'