MCP Server开发教程:从零写一个自己的AI工具
🩺 摘要
MCP协议知道了,但怎么开发一个自己的MCP Server?
📝 详情
MCP Server从零搭建:把你的系统变成AI可调用的API
MCP是什么
MCP(Model Context Protocol)是由Anthropic提出的开放协议,类似于AI模型的"USB接口"——写一个MCP Server,任何支持MCP的AI(Claude、GPT、DeepSeek等)都能自动发现并调用你的工具。
快速开始:完整代码
第一步:安装MCP SDK
pip install mcp # Python MCP SDK,支持stdio和HTTP传输
第二步:写一个MCP Server
# weather_mcp_server.py
from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationOptions
import mcp.server.stdio
import mcp.types as types
import httpx
from typing import Any
# 创建MCP Server实例
server = Server("weather-server")
# 用装饰器注册工具——这就是AI能调用的接口
@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
return [
types.Tool(
name="get_weather",
description="查询指定城市的天气",
inputSchema={
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名,如北京、上海"},
"units": {"type": "string", "enum": ["celsius", "fahrenheit"]}
},
"required": ["city"]
}
),
types.Tool(
name="get_forecast",
description="获取未来7天天气预报",
inputSchema={
"type": "object",
"properties": {
"city": {"type": "string"},
"days": {"type": "integer", "maximum": 7}
},
"required": ["city"]
}
)
]
@server.call_tool()
async def handle_call_tool(name: str, arguments: dict[str, Any]) -> list[types.TextContent]:
if name == "get_weather":
city = arguments["city"]
async with httpx.AsyncClient() as client:
resp = await client.get(f"https://api.weather.com/v1/{city}")
data = resp.json()
return [types.TextContent(type="text", text=json.dumps(data, ensure_ascii=False))]
elif name == "get_forecast":
city = arguments["city"]
days = arguments.get("days", 3)
# ... 预报逻辑
return [types.TextContent(type="text", text=f"{city}未来{days}天预报...")]
raise ValueError(f"未知工具: {name}")
# 启动Server(stdio模式,与AI客户端通过标准输入输出通信)
async def main():
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
await server.run(
read_stream, write_stream,
InitializationOptions(server_name="weather-server")
)
if __name__ == "__main__":
import asyncio
asyncio.run(main())
第三步:配置AI客户端
// Claude Desktop 配置(claude_desktop_config.json)
{
"mcpServers": {
"weather-server": {
"command": "python",
"args": ["weather_mcp_server.py"]
}
}
}
// 支持HTTP模式的配置
{
"mcpServers": {
"weather-server": {
"url": "http://localhost:8080/mcp"
}
}
}
进阶:MCP Server的最佳实践
# mcp_best_practice.py — 带认证和错误处理
from mcp.server import Server
import mcp.types as types
server = Server("enterprise-server")
TOKEN = "your-service-token"
@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list[types.TextContent]:
try:
# 参数校验
if "auth_token" not in arguments or arguments["auth_token"] != TOKEN:
return [types.TextContent(type="text", text="{\"error\": \"未授权\"}")]
if name == "query_database":
sql = arguments.get("sql", "")
# SQL注入防护
if any(kw in sql.upper() for kw in ["DROP", "DELETE", "UPDATE"]):
return [types.TextContent(type="text", text="{\"error\": \"只允许查询操作\"}")]
result = execute_readonly_query(sql)
return [types.TextContent(type="text", text=json.dumps(result, ensure_ascii=False))]
elif name == "search_knowledge_base":
query = arguments.get("query", "")
results = vector_db.search(query, top_k=5)
return [types.TextContent(type="text", text=json.dumps(results, ensure_ascii=False))]
except Exception as e:
return [types.TextContent(type="text", text=json.dumps({"error": str(e)}))]
raise ValueError(f"未知工具: {name}")
已上线的MCP Server案例
| 类型 | 示例 | 被集成的AI数 |
|---|---|---|
| 数据库查询 | MySQL/PostgreSQL查询接口 | 50+ |
| 搜索 | Elasticsearch/Meilisearch | 100+ |
| 文件操作 | Google Drive/Dropbox | 30+ |
| 业务系统 | Salesforce/SAP | 20+ |
| 自定义 | 企业内部系统API | 5000+(GitHub) |
避坑指南
- 工具描述要详细:AI根据description决定调用哪个工具,描述越具体AI用得越准
- 参数要有默认值:尽量给每个参数设定合理的默认值,降低AI调用的门槛
- 返回值要结构化:用JSON而不是纯文本,方便AI解析
- 部署首选stdio:本地开发用stdio最简单;线上用HTTP模式支持多客户端
💬 评论 (0)