MCP 实战：从零搭建 Model Context Protocol Server，让 AI 连接你的工具
TL;DR：MCP（Model Context Protocol）是 Anthropic 推出的开放协议，让 AI 模型能标准化地连接外部工具和数据源。本文从协议原理讲起，手把手带你用 Python 搭建一个 MCP Server，并接入 Claude / Cursor 使用。
1. 为什么需要 MCP
过去让 AI 调用工具，每个应用都要自己实现一遍集成：
- 想让 Claude 读数据库 → 自己写一套工具调用逻辑
- 想让 Cursor 访问文件系统 → 再写一套
- 想让自研 Agent 调 API → 又写一套
结果就是N 个模型 × M 个工具 = N×M 个集成，每个都是定制化代码。
MCP 的出现解决了这个问题：
MCP 架构
┌──────────────┐ MCP 协议 ┌──────────────┐ │ AI 应用 │ ←────────────→ │ MCP Server │ │ (Claude/Cursor│ (stdio/HTTP) │ (你的工具) │ │ /Agent) │ │ - 数据库 │ └──────────────┘ │ - 文件系统 │ │ - API │ └──────────────┘
模型只需要支持 MCP 协议，就能连接任意 MCP Server。一次开发，处处可用。
2. MCP 的核心概念
| 概念 | 说明 |
|---|---|
| Host | AI 应用本身（Claude Desktop、Cursor、自研 Agent） |
| Client | Host 内部的 MCP 客户端，负责和 Server 通信 |
| Server | 提供能力的服务（工具、资源、提示词模板） |
| Tool | 可调用的函数（如查询数据库、发送邮件） |
| Resource | 可读取的数据（如文件、API 响应） |
| Prompt | 预定义的提示词模板 |
3. 搭建第一个 MCP Server
3.1 环境准备
bash
pip install mcp pip install uvicorn # 如果用 HTTP 模式 # 验证安装 python -c "import mcp; print('MCP installed')"3.2 用 FastMCP 写一个简单 Server
Python - mcp_server.py
from mcp.server.fastmcp import FastMCP import sqlite3 import os # 创建 MCP Server mcp = FastMCP("company-db-server") DB_PATH = "./company.db" def get_db_connection(): conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row return conn # ========== Tool 1: 查询员工 ========== def query_employee(name: str) -> str: """根据姓名查询员工信息 Args: name: 员工姓名（支持模糊匹配） """ conn = get_db_connection() cursor = conn.cursor() cursor.execute( "SELECT * FROM employees WHERE name LIKE ?", (f"%{name}%",) ) rows = cursor.fetchall() conn.close() if not rows: return f"未找到匹配 '{name}' 的员工" result = [] for row in rows: result.append( f"姓名: {row['name']}, 部门: {row['department']}, " f"职位: {row['title']}, 入职: {row['hire_date']}" ) return "\n".join(result) # ========== Tool 2: 统计部门人数 ========== def count_department(department: str) -> str: """统计某个部门的员工人数 Args: department: 部门名称 """ conn = get_db_connection() cursor = conn.cursor() cursor.execute( "SELECT COUNT(*) as count FROM employees WHERE department = ?", (department,) ) count = cursor.fetchone()["count"] conn.close() return f"部门 '{department}' 共有 {count} 名员工" # ========== Resource: 数据库 schema ========== "schema://employees") def get_schema() -> str: """返回 employees 表的字段定义""" return """employees 表结构： - id: 主键 - name: 姓名 - department: 部门 - title: 职位 - salary: 薪资 - hire_date: 入职日期""" # 启动 Server（stdio 模式） if __name__ == "__main__": mcp.run()3.3 在 Claude Desktop 中配置
claude_desktop_config.json
{ "mcpServers": { "company-db": { "command": "python", "args": ["/path/to/mcp_server.py"] } } }重启 Claude Desktop 后，它会自动发现你的 MCP Server，你就能直接对话：
你：「公司技术部有多少人？」
Claude：（自动调用count_department("技术部")）→ 「技术部共有 15 名员工」
4. 进阶：带鉴权的 MCP Server
生产环境需要安全控制。给 MCP Server 加上 API Key 鉴权：
Python - 带鉴权的 MCP Server
from mcp.server.fastmcp import FastMCP from mcp.server import Server from mcp.types import Tool, TextContent import os EXPECTED_TOKEN = os.getenv("MCP_TOKEN", "default-secret") # 自定义 Server，拦截请求做鉴权 server = Server("secure-db-server") @server.list_tools() async def list_tools() -> list[Tool]: return [ Tool( name="query_employee", description="查询员工信息", inputSchema={ "type": "object", "properties": { "name": {"type": "string"} } } ) ] @server.call_tool() async def call_tool(name: str, arguments: dict) -> list[TextContent]: # 这里简化：真实场景从请求头读取 token if name == "query_employee": result = query_employee(arguments["name"]) return [TextContent(type="text", text=result)] return [TextContent(type="text", text="Unknown tool")] # 启动 if __name__ == "__main__": import mcp.server.stdio async def main(): async with mcp.server.stdio.stdio_server() as (read, write): await server.run(read, write, server.create_initialization_options()) asyncio.run(main())5. 实战：构建一个 GitHub MCP Server
让 AI 能查 Issue、建 PR、读代码：
Python - github_mcp_server.py
from mcp.server.fastmcp import FastMCP import requests import os mcp = FastMCP("github-server") GITHUB_TOKEN = os.getenv("GITHUB_TOKEN") BASE_URL = "https://api.github.com" HEADERS = { "Authorization": f"token {GITHUB_TOKEN}", "Accept": "application/vnd.github.v3+json" } def list_open_issues(repo: str, limit: int = 10) -> str: """列出仓库的开放 Issue Args: repo: 格式为 "owner/repo" limit: 返回数量限制 """ response = requests.get( f"{BASE_URL}/repos/{repo}/issues", headers=HEADERS, params={"state": "open", "per_page": limit} ) issues = response.json() result = [] for issue in issues[:limit]: result.append(f"#{issue['number']} {issue['title']} ({issue['html_url']})") return "\n".join(result) if result else "没有开放 Issue" def create_issue(repo: str, title: str, body: str = "") -> str: """创建新的 GitHub Issue Args: repo: 格式为 "owner/repo" title: Issue 标题 body: Issue 内容 """ response = requests.post( f"{BASE_URL}/repos/{repo}/issues", headers=HEADERS, json={"title": title, "body": body} ) if response.status_code == 201: data = response.json() return f"✅ 已创建 Issue #{data['number']}: {data['html_url']}" return f"❌ 创建失败: {response.status_code} {response.text}" if __name__ == "__main__": mcp.run()6. MCP vs Function Calling
| 维度 | MCP | Function Calling |
|---|---|---|
| 定位 | 标准化协议（连接方式） | 模型能力（调用方式） |
| 复用性 | 一次开发，多模型使用 | 每个模型各自实现 |
| 生态 | 跨应用共享 Server | 绑定单一应用 |
| 复杂度 | 需要 Server 进程 | 直接传函数定义 |
| 适用 | 工具生态建设 | 简单工具调用 |
⚠️ 注意：MCP 和 Function Calling 不是对立的。MCP Server 底层也是用 Function Calling 实现工具调用，只是封装了一层标准化协议，让工具可以跨模型、跨应用复用。
7. 总结
MCP 解决了 AI 工具集成的「N×M 问题」：
- 过去：每个 AI 应用自己写工具集成，重复劳动
- 现在：写一次 MCP Server，所有支持 MCP 的 AI 应用都能用
适合用 MCP 的场景：
- 你有一套内部工具/API，想让多个 AI 应用都能调用
- 你做 Agent 平台，想让用户自己贡献工具
- 你需要统一的安全/鉴权层管理工具访问
入门建议：先用 FastMCP 写一个简单 Server（如查询数据库），接入 Claude Desktop 体验。熟悉后再做鉴权、HTTP 模式、多工具编排。
如果对你有帮助，欢迎在评论区聊聊你用 MCP 踩过的坑。