大模型 MCP 实战：5分钟搞懂 Model Context Protocol，让 AI 调用工具像呼吸一样自然

Anthropic 发布的 Model Context Protocol（MCP）正在成为 AI 工具调用的新标准。本文从原理到实战，完整解析 MCP 的前世今生，以及如何用它快速扩展大模型的工具调用能力。

---

背景：AI 工具调用的”战国时代”

在 MCP 出现之前，大模型想调用外部工具有多痛苦？

• Function Calling：每个模型有自己的格式，OpenAI 的和 Anthropic 的不兼容
• 插件生态：GPTs 的插件、Claude 的工具，每个平台各玩各的
• 重复造轮子：每接入一个新工具，都要写一套新的适配代码

举个例子，你想让大模型同时调用 GitHub API、Slack 和内部数据库：

传统方案：
GPT-4 → 你写的 adapter → GitHub API
Claude → 你写的 adapter → Slack  
DeepSeek → 你写的 adapter → 内部数据库

MCP 方案：
任意大模型 → MCP Client → MCP Server (GitHub)
                           MCP Server (Slack)
                           MCP Server (内部DB)

MCP 就像 USB 接口——不管你用什么电脑（模型），只要是 USB 接口的设备（工具），插上就能用。

---

MCP 核心架构解析

三组件模型

组件	职责	类比
MCP Host	运行 AI 应用的宿主环境（Claude Desktop、Cursor 等）	电脑主机
MCP Client	在 Host 内运行，与 Server 保持 1:1 连接	USB 控制器
MCP Server	暴露特定工具/资源的独立进程	USB 设备

通信协议

MCP 使用 JSON-RPC 2.0 作为通信协议，支持两种传输方式：

1. stdio：通过标准输入输出通信，适合本地运行的 Server
2. HTTP + SSE：适合远程部署的 Server（企业级场景）

---

快速上手：3步运行你的第一个 MCP Server

第1步：安装 Claude Desktop + MCP SDK

# 安装 Node.js 环境
# macOS
brew install node

# Windows (使用 winget)
winget install OpenJS.NodeJS.LTS

# 全局安装 MCP SDK
npm install -g @anthropic-ai/claude-mcp-cli

第2步：编写一个文件搜索 MCP Server

# file_search_server.py
# 使用 Python MCP SDK 实现文件搜索工具
import os
import json
from pathlib import Path
from typing import Any
from mcp.server.fastmcp import FastMCP

# 初始化 MCP Server
mcp = FastMCP("file-search", timeout=30)

# 定义资源（AI 可以主动读取的数据）
@mcp.resource("workspace://files")
def list_workspace_files() -> str:
    """列出工作区所有文件"""
    files = []
    for root, dirs, filenames in os.walk("."):
        # 忽略隐藏文件和 node_modules
        dirs[:] = [d for d in dirs if not d.startswith(".")]
        for f in filenames:
            if not f.startswith("."):
                files.append(os.path.join(root, f))
    return json.dumps(files, indent=2)


@mcp.resource("file://{path}")
def read_file_content(path: str) -> str:
    """读取指定文件内容（动态路径）"""
    try:
        file_path = Path(path).resolve()
        if not file_path.exists():
            return f"Error: File not found: {path}"
        return file_path.read_text(encoding="utf-8")
    except Exception as e:
        return f"Error reading {path}: {str(e)}"


# 定义工具（AI 可以调用的函数）
@mcp.tool()
def search_in_files(
    directory: str = ".",
    pattern: str = "",
    file_type: str = "*",
    max_results: int = 50
) -> dict[str, Any]:
    """
    在指定目录下搜索包含关键词的文件

    Args:
        directory: 搜索目录，默认为当前目录
        pattern: 搜索的关键词
        file_type: 文件类型过滤，如 "py", "md", "json"
        max_results: 最大返回数量

    Returns:
        包含匹配结果的字典
    """
    results = []
    search_path = Path(directory).resolve()

    if not search_path.exists():
        return {"error": f"Directory not found: {directory}"}

    # 根据文件类型构建 glob 模式
    glob_pattern = f"**/*.{file_type}" if file_type != "*" else "**/*"

    for i, file_path in enumerate(search_path.glob(glob_pattern)):
        if i >= max_results:
            break

        # 跳过隐藏文件和二进制文件
        if any(p.startswith(".") for p in file_path.parts):
            continue

        try:
            content = file_path.read_text(encoding="utf-8", errors="ignore")
            if pattern.lower() in content.lower():
                # 找到匹配行
                matches = [
                    line.strip() for line in content.split("\n")
                    if pattern.lower() in line.lower()
                ]
                results.append({
                    "file": str(file_path.relative_to(search_path)),
                    "total_matches": len(matches),
                    "preview": matches[:3]  # 只返回前3条匹配
                })
        except Exception:
            continue  # 跳过无法读取的文件

    return {
        "search_directory": str(search_path),
        "pattern": pattern,
        "total_files_with_matches": len(results),
        "results": results[:20]  # 最多返回20个文件
    }


@mcp.tool()
def grep_file(
    file_path: str,
    pattern: str,
    context_lines: int = 2
) -> str:
    """
    在单个文件中搜索并返回上下文

    Args:
        file_path: 文件路径
        pattern: 搜索模式（支持正则）
        context_lines: 匹配行前后的行数

    Returns:
        格式化的搜索结果
    """
    import re

    try:
        path = Path(file_path)
        if not path.exists():
            return f"Error: File not found: {file_path}"

        content = path.read_text(encoding="utf-8", errors="ignore")
        lines = content.split("\n")

        matches = []
        for i, line in enumerate(lines):
            if re.search(pattern, line, re.IGNORECASE):
                # 获取上下文行
                start = max(0, i - context_lines)
                end = min(len(lines), i + context_lines + 1)

                context = []
                for j in range(start, end):
                    prefix = ">>> " if j == i else "    "
                    context.append(f"{prefix}{j+1}: {lines[j]}")

                matches.append("\n".join(context))

        if not matches:
            return f"No matches found for '{pattern}' in {file_path}"

        return f"Found {len(matches)} matches in {file_path}:\n\n" + "\n---\n".join(matches)

    except Exception as e:
        return f"Error: {str(e)}"


# 启动 MCP Server
if __name__ == "__main__":
    print("🚀 启动文件搜索 MCP Server...")
    mcp.run(transport="stdio")

第3步：配置 Claude Desktop 使用这个 Server

// macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
// Windows: %APPDATA%\Claude\claude_desktop_config.json

{
  "mcpServers": {
    "file-search": {
      "command": "python",
      "args": [
        "/path/to/file_search_server.py"
      ],
      "env": {
        "PYTHONPATH": "/path/to/site-packages"
      }
    }
  }
}

重启 Claude Desktop，你就可以这样对话了：

你：在当前目录下搜索所有包含 "async def" 的 Python 文件

Claude（自动调用 search_in_files 工具）：
找到了 5 个文件，包含 "async def" 的函数定义如下：

1. server.py (3 matches)
   - line 45: async def handle_request()
   - line 78: async def process_stream()
   - line 120: async def cleanup()

2. client.py (2 matches)
   ...

---

进阶实战：连接真实 API（MCP + GitHub）

# github_mcp_server.py
import os
from github import Github
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("github-tools")

# 初始化 GitHub 客户端
github_token = os.environ.get("GITHUB_TOKEN")
gh = Github(github_token)


@mcp.tool()
def create_issue(
    repo: str,
    title: str,
    body: str = "",
    labels: list[str] = None
) -> dict:
    """在 GitHub 仓库创建一个 Issue"""
    repository = gh.get_repo(repo)
    issue = repository.create_issue(
        title=title,
        body=body,
        labels=labels or []
    )
    return {
        "number": issue.number,
        "title": issue.title,
        "url": issue.html_url,
        "state": issue.state
    }


@mcp.tool()
def search_code(
    query: str,
    language: str = None,
    max_results: int = 10
) -> list[dict]:
    """搜索 GitHub 代码"""
    from github.GithubException import RateLimitExceededException

    try:
        results = gh.search_code(
            query=query,
            language=language,
            sort="indexed",
            order="desc"
        )

        return [
            {
                "file": r.path,
                "repo": r.repository.full_name,
                "url": r.html_url,
                "preview": r.text_matches[:3] if hasattr(r, 'text_matches') else []
            }
            for r in results[:max_results]
        ]
    except RateLimitExceededException:
        return [{"error": "GitHub API rate limit exceeded. Try again later."}]


@mcp.tool()
def get_repo_stats(repo: str) -> dict:
    """获取仓库统计信息"""
    repository = gh.get_repo(repo)

    return {
        "name": repository.full_name,
        "stars": repository.stargazers_count,
        "forks": repository.forks_count,
        "open_issues": repository.open_issues_count,
        "language": repository.language,
        "description": repository.description,
        "latest_release": repository.last_release.tag_name if repository.last_release else None,
        "contributors_count": repository.get_contributors().totalCount
    }


@mcp.resource("repo://{owner}/{name}/readme")
def get_readme(owner: str, name: str) -> str:
    """动态获取任意仓库的 README"""
    repo = gh.get_repo(f"{owner}/{name}")
    return repo.get_readme().decoded_content.decode("utf-8")


if __name__ == "__main__":
    print("🚀 启动 GitHub MCP Server...")
    mcp.run(transport="stdio")

配置后，Claude 可以这样工作：

你：帮我看看 apache/airflow 仓库的 star 数，最近一次 release 版本，以及创建者是谁

Claude（自动调用工具）：
让我查询一下...

📊 apache/airflow 统计信息：

- ⭐ Stars: 35,000+
- 🍴 Forks: 17,000+
- 📝 语言: Python
- 📖 最新 Release: v2.9.0 (2024-04)
- 👤 创建者: apache

看起来是个非常活跃的 Python 工作流平台！

---

MCP vs Function Calling vs 插件生态

特性	MCP	Function Calling	GPTs 插件
标准化程度	✅ CNCF 标准	❌ 各厂商自定义	❌ 平台锁定
工具发现机制	✅ 自动发现 + schema	❌ 手动注册	❌ 需 OpenAPI 规范
上下文保持	✅ 有状态会话	❌ 无状态	❌ 有限
生态丰富度	快速成长中	成熟但分散	丰富但封闭
适合场景	开发者工具链	简单 API 调用	C端消费者应用
学习成本	中等	低	高

---

生产环境部署：MCP Server 高可用方案

使用 Docker 部署远程 MCP Server

# mcp-server-docker-compose.yml
version: '3.8'

services:
  github-server:
    build: ./github-mcp-server
    environment:
      - GITHUB_TOKEN=${GITHUB_TOKEN}
    deploy:
      replicas: 2
      resources:
        limits:
          cpus: '0.5'
          memory: 256M

  file-search-server:
    build: ./file-search-mcp-server
    volumes:
      - /data/workspaces:/workspaces:ro  # 只读挂载
    deploy:
      replicas: 2

  # Nginx 作为 MCP HTTP+SSE 的反向代理
  mcp-proxy:
    image: nginx:alpine
    ports:
      - "8080:8080"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf:ro
    depends_on:
      - github-server
      - file-search-server

SSE 传输模式的 MCP Server 实现

# 使用 FastAPI 提供 HTTP+SSE 传输
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse, StreamingResponse
from sse_starlette.sse import EventSourceResponse
import asyncio
import json

app = FastAPI()

# MCP Server 实例（省略初始化代码）
from github_mcp_server import mcp

@app.post("/mcp")
async def mcp_endpoint(request: Request):
    """处理 MCP JSON-RPC 请求"""
    body = await request.json()
    method = body.get("method")
    params = body.get("params", {})
    msg_id = body.get("id")

    # 处理不同的 MCP 方法
    if method == "tools/list":
        tools = await mcp.list_tools()
        return JSONResponse({
            "jsonrpc": "2.0",
            "id": msg_id,
            "result": {"tools": tools}
        })

    elif method == "tools/call":
        tool_name = params.get("name")
        arguments = params.get("arguments", {})
        result = await mcp.call_tool(tool_name, arguments)
        return JSONResponse({
            "jsonrpc": "2.0",
            "id": msg_id,
            "result": {"content": [{"type": "text", "text": json.dumps(result)}]}
        })

    return JSONResponse({
        "jsonrpc": "2.0",
        "id": msg_id,
        "error": {"code": -32601, "message": "Method not found"}
    })


@app.get("/mcp/stream")
async def mcp_stream(request: Request):
    """SSE 长连接，用于推送日志/进度"""
    async def event_generator():
        yield {"event": "message", "data": json.dumps({"status": "connected"})}

        # 可以在这里推送长时任务的进度
        for i in range(10):
            await asyncio.sleep(1)
            yield {"event": "progress", "data": json.dumps({"percent": i * 10})}

    return EventSourceResponse(event_generator())

---

总结：MCP 的价值与局限

什么时候用 MCP？

✅ 推荐使用 MCP：
– 需要让多个 AI 模型共用同一套工具
– 团队内部有大量可复用的工具/API
– 想要构建 AI 原生的开发工具链
– 可观测性要求高的生产环境

❌ MCP 不适合的场景：
– 只需要简单调用 1-2 个 API（直接写 Function Calling 更简单）
– 工具逻辑非常固定，不需要 AI 判断（直接硬编码更高效）
– 已有成熟的 LangChain/LlamaIndex 体系（迁移成本高）

---

👤 作者简介

一枚在大中原腹地（河南）卖公有云的从业者，主营腾讯云/阿里云/火山云，曾踩坑无数，现专注AI大模型应用落地。关注公众号「AI热点日报」，围观AI前沿动态~

博客：y1cloud.cn

MCP 生态正在快速发展，你最期待 MCP 接入什么工具/服务？GitHub、Notion、还是内部数据库？评论区聊聊！