MCP 实战：从零搭建 Model Context Protocol Server，让 AI 连接你的工具

路从脚起

45人浏览 · 2026-07-07 22:38:44

路从脚起 · 2026-07-07 22:38:44 发布

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 进程	直接传函数定义
适用	工具生态建设	简单工具调用