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 的核心概念

概念说明
HostAI 应用本身(Claude Desktop、Cursor、自研 Agent)
ClientHost 内部的 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

维度MCPFunction 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 踩过的坑。