Python构建MCP服务器：从工具封装到AI集成的全流程实践

2025-08-21 1367

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

简介： MCP协议为AI提供标准化工具调用接口，助力模型高效操作现实世界。

免费获取编程教程：https://pan.quark.cn/s/876976d33a34
一、MCP协议：AI与工具的"USB-C接口"
想象你正在用AI助手处理工作：需要查询天气时，AI突然弹出"我需要调用天气API"的提示；处理Excel数据时，它又卡在"如何读取CSV文件"的步骤。传统AI的局限性在于，它像一台没有外设的电脑——能思考却无法操作现实世界。
探秘代理IP并发连接数限制的那点事 (54).png

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调用你的工具

配置Claude Desktop
编辑配置文件 claude_desktop_config.json：

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

自然语言调用示例
当用户在Claude中输入：

"查询北京今天的天气"

AI内部处理流程：

发现可用的 get_weather 工具
将自然语言转换为工具调用：

{
"toolName": "get_weather",
"arguments": {"city": "北京"}
}
执行工具并返回结果：

{
"temperature": 28,
"humidity": 65,
"condition": "多云"
}
将结果生成自然语言回复：
"北京今天气温28℃，湿度65%，天气多云"

调试技巧
使用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"}}'

八、未来展望：MCP生态演进方向
工具市场：类似PyPI的MCP工具仓库，实现"一键安装"工具
协议扩展：支持gRPC、WebSocket等更多通信协议
智能路由：根据请求内容自动选择最优工具
边缘计算：在IoT设备上部署轻量级MCP服务器
MCP协议正在重塑AI开发范式——它让大模型从"封闭的大脑"进化为"可连接万物的神经系统"。无论是个人开发者快速扩展AI能力，还是企业整合遗留系统，MCP都提供了标准化解决方案。当工具调用变得像呼吸一样自然，AI才能真正成为生产力的延伸。

Python构建MCP服务器：从工具封装到AI集成的全流程实践

使用uv工具快速初始化项目（推荐）

或 .venv\Scripts\activate (Windows)

基础版（适合个人开发）

企业版（需处理高并发）

创建test_install.py

weather_tool.py

db_tool.py

server_stdio.py

server_sse.py

converter.py

示例：转换天气API

使用Vault管理密钥

使用Gunicorn部署FastMCP应用

tools/stock_api.py

server.py

安装依赖

启动服务器

测试工具调用

热门文章

最新文章

相关课程

相关电子书

相关实验场景

推荐镜像

探索云世界

热门

云计算

大数据

云原生

人工智能

数据库

开发与运维

活动广场

任务中心

训练营

直播

乘风者计划

下载

镜像站

技术资料

Python构建MCP服务器：从工具封装到AI集成的全流程实践

使用uv工具快速初始化项目（推荐）

或 .venv\Scripts\activate (Windows)

基础版（适合个人开发）

企业版（需处理高并发）

创建test_install.py

weather_tool.py

db_tool.py

server_stdio.py

server_sse.py

converter.py

示例：转换天气API

使用Vault管理密钥

使用Gunicorn部署FastMCP应用

tools/stock_api.py

server.py

安装依赖

启动服务器

测试工具调用

热门文章

最新文章

相关课程

相关电子书

相关实验场景

推荐镜像