MCP协议实战：零基础用Python开发首个AI工具

大型语言模型正在获得执行能力——MCP协议让AI应用不仅能生成文本，还能真正运行代码、调用API、操作文件。本文从零开始，完整演示一个MCP Server的搭建全过程。

一、MCP协议是什么？

1.1 核心概念

MCP（模型上下文协议）定义了一套标准化规范，使AI应用（Host）能通过统一接口访问外部工具和数据。这套架构初看复杂，但拆解后逻辑清晰：AI应用作为主机，中间层是MCP客户端，外层是MCP服务端，三者通过标准通信管道协作。

从架构全景图可见，MCP服务端主要提供三类能力：Tools（工具调用）、Resources（数据读取）和Prompts（提示模板）。Host内部的MCP客户端充当翻译器，在AI应用和服务端之间传递消息。

1.2 MCP三大核心能力

1.3 通信方式

目前支持两种传输方式，选择依据取决于应用场景。本地开发推荐stdio模式：AI应用将MCP Server作为子进程启动，通过标准输入输出通信，实现低延迟集成。远程部署则采用SSE（Server-Sent Events），即通过HTTP进行网络通信。两者各有优劣：stdio适合快速原型，SSE适用于生产环境的多客户端并发访问。

方式一: stdio（标准输入输出）
┌──────────┐ stdin/stdout   ┌──────────┐
│ AI 应用  │◄──────────────►│MCP Server│
│ (Host)   │                │(子进程)  │
└──────────┘                └──────────┘

方式二: SSE（Server-Sent Events）
┌──────────┐ HTTP + SSE     ┌──────────┐
│ AI 应用  │◄──────────────►│MCP Server│
│ (Host)   │ (网络通信)     │(远程服务)│
└──────────┘                └──────────┘

二、开发环境搭建

2.1 安装依赖

执行命令：pip install "mcp[cli]" httpx，一行完成依赖安装。httpx作为异步HTTP客户端，功能优于requests，后续天气查询将使用它。

2.2 验证安装

运行 mcp --version，若输出版本号则表示安装成功。

2.3 项目结构

推荐采用以下目录组织方式：

my-mcp-server/
├── server.py              # MCP Server主文件
├── tools/
│   ├── weather.py         # 天气查询工具
│   ├── database.py        # 数据库查询工具
│   └── calculator.py      # 计算器工具
├── resources/
│   └── system_info.py     # 系统信息资源
├── prompts/
│   └── code_review.py     # 代码审查提示模板
├── pyproject.toml         # 项目配置
└── README.md

按功能模块拆分后，每个文件职责明确，后期维护无需面对数千行的臃肿代码库。

三、第一个MCP Server（快速上手）

3.1 最简实现

先编写一个极简示例，快速体验MCP的核心能力：

# server.py
from mcp.server.fastmcp import FastMCP

# 创建MCP Server实例
mcp = FastMCP(
    name="my-first-mcp-server",
    version="1.0.0",
)

@mcp.tool()
def add(a: int, b: int) -> int:
    """两个数字相加"""
    return a + b

@mcp.tool()
def get_current_time(timezone: str = "Asia/Shanghai") -> str:
    """获取当前时间
    Args:
        timezone: 时区名称，默认为Asia/Shanghai
    """
    from datetime import datetime
    import pytz
    tz = pytz.timezone(timezone)
    now = datetime.now(tz)
    return now.strftime("%Y-%m-%d %H:%M:%S %Z")

if __name__ == "__main__":
    mcp.run()

注意 @mcp.tool() 装饰器——这是将普通Python函数转化为AI可调用工具的关键。参数类型标注和docstring并非摆设，AI正是依赖这些信息来理解工具的调用方式。

3.2 运行测试

提供两种启动方式：mcp dev server.py 会启动可视化调试工具（MCP Inspector），在浏览器中即可测试工具。直接执行 python server.py 也可正常启动。

四、实战：开发完整的企业工具服务

下面构建一个实用的MCP Server，涵盖天气查询、数据库操作、文件管理等功能。

4.1 天气查询工具

调用OpenWeatherMap API，支持当前天气和天气预报：

# tools/weather.py
import httpx
from mcp.server.fastmcp import tool

BASE_URL = "https://api.openweathermap.org/data/2.5"

@tool()
async def get_weather(city: str, api_key: str = "") -> str:
    """查询指定城市的当前天气信息
    Args:
        city: 城市名称（中文或英文），如"北京"或"Beijing"
        api_key: OpenWeatherMap API Key（可选，默认使用内置测试key）
    """
    params = {
        "q": city,
        "appid": api_key,
        "units": "metric",
        "lang": "zh_cn",
    }
    async with httpx.AsyncClient() as client:
        resp = await client.get(f"{BASE_URL}/weather", params=params)
        data = resp.json()

    if resp.status_code != 200:
        return f"查询失败: {data.get('message', '未知错误')}"

    return (
        f"【{data['name']} 天气】\n"
        f"天气: {data['weather'][0]['description']}\n"
        f"温度: {data['main']['temp']}°C (体感 {data['main']['feels_like']}°C)\n"
        f"湿度: {data['main']['humidity']}%\n"
        f"风速: {data['wind']['speed']} m/s"
    )

@tool()
async def get_forecast(city: str, days: int = 3, api_key: str = "") -> str:
    """查询未来几天的天气预报
    Args:
        city: 城市名称
        days: 预报天数（最多5天）
        api_key: OpenWeatherMap API Key
    """
    params = {
        "q": city,
        "appid": api_key,
        "units": "metric",
        "lang": "zh_cn",
        "cnt": min(days, 5) * 8,  # 每天约8个数据点
    }
    async with httpx.AsyncClient() as client:
        resp = await client.get(f"{BASE_URL}/forecast", params=params)
        data = resp.json()

    forecasts = []
    for item in data["list"][::8]:  # 每天取一个数据点
        forecasts.append(
            f"{item['dt_txt'][:10]}: "
            f"{item['weather'][0]['description']}, "
            f"{item['main']['temp']}°C"
        )

    return f"【{data['city']['name']} 未来天气预报】\n" + "\n".join(forecasts)

4.2 SQLite数据库查询工具

工具设计注重安全性——仅允许SELECT操作，防止AI意外执行危险指令：

# tools/database.py
import sqlite3
from pathlib import Path
from mcp.server.fastmcp import tool

DB_PATH = Path("./data/app.db")

def _get_connection():
    """获取数据库连接"""
    DB_PATH.parent.mkdir(exist_ok=True)
    return sqlite3.connect(str(DB_PATH))

@tool()
def query_database(sql: str) -> str:
    """执行SQL查询语句（仅支持SELECT）
    Args:
        sql: SQL查询语句，仅允许SELECT操作
    """
    # 安全检查：只允许SELECT
    normalized = sql.strip().upper()
    if not normalized.startswith("SELECT"):
        return "错误：仅支持SELECT查询，不允许修改数据"
    
    try:
        conn = _get_connection()
        cursor = conn.execute(sql)
        columns = [desc[0] for desc in cursor.description]
        rows = cursor.fetchall()
        conn.close()

        if not rows:
            return "查询结果为空"
        
        header = "| " + " | ".join(columns) + " |"
        separator = "| " + " | ".join(["---"] * len(columns)) + " |"
        data_rows = []
        for row in rows[:50]:  # 最多返回50行
            data_rows.append("| " + " | ".join(str(v) for v in row) + " |")
        table = "\n".join([header, separator] + data_rows)
        return f"查询返回 {len(rows)} 行:\n{table}"
    
    except Exception as e:
        return f"查询出错: {str(e)}"

@tool()
def list_tables() -> str:
    """列出数据库中的所有表及其结构"""
    conn = _get_connection()
    cursor = conn.execute("SELECT name FROM sqlite_master WHERE type='table'")
    tables = cursor.fetchall()
    conn.close()

    if not tables:
        return "数据库为空，暂无表"
    
    result = []
    for (table_name,) in tables:
        result.append(f"- **{table_name}**")
    
    return "数据库中的表:\n" + "\n".join(result)

4.3 文件管理工具

文件操作需严格防范路径穿越攻击，因此实现了一个沙箱限制：

# tools/file_manager.py
import os
from pathlib import Path
from mcp.server.fastmcp import tool

SAFE_DIR = Path("./data/workspace")  # 安全沙箱目录

def _safe_path(filepath: str) -> Path:
    """确保文件路径在安全目录内"""
    full_path = (SAFE_DIR / filepath).resolve()
    if not str(full_path).startswith(str(SAFE_DIR.resolve())):
        raise ValueError("路径超出安全范围")
    return full_path

@tool()
def read_file(filepath: str) -> str:
    """读取文件内容
    Args:
        filepath: 相对于工作目录的文件路径
    """
    path = _safe_path(filepath)
    if not path.exists():
        return f"文件不存在: {filepath}"
    return path.read_text(encoding="utf-8")

@tool()
def write_file(filepath: str, content: str) -> str:
    """写入文件内容
    Args:
        filepath: 相对于工作目录的文件路径
        content: 要写入的内容
    """
    path = _safe_path(filepath)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(content, encoding="utf-8")
    return f"文件写入成功: {filepath} ({len(content)} 字符)"

@tool()
def list_files(directory: str = ".") -> str:
    """列出目录下的文件
    Args:
        directory: 相对于工作目录的目录路径
    """
    path = _safe_path(directory)
    if not path.exists():
        return f"目录不存在: {directory}"
    
    items = []
    for item in sorted(path.iterdir()):
        if item.is_dir():
            items.append(f"???? {item.name}/")
        else:
            size = item.stat().st_size
            items.append(f"???? {item.name} ({size} bytes)")
    
    return "\n".join(items) if items else "目录为空"

4.4 定义Resources（数据资源）

Resources是AI可直接读取的数据源，相当于只读接口：

# resources/system_info.py
import platform
import psutil
from mcp.server.fastmcp import resource

@resource("system://info")
def get_system_info() -> str:
    """获取系统信息"""
    return f"""
系统信息:
- 操作系统: {platform.system()} {platform.release()}
- Python: {platform.python_version()}
- CPU核心: {psutil.cpu_count()}
- 内存总量: {psutil.virtual_memory().total // (1024**3)} GB
- 磁盘使用: {psutil.disk_usage('/').percent}%
"""

@resource("system://processes")
def get_top_processes() -> str:
    """获取占用资源最多的前10个进程"""
    procs = []
    for p in psutil.process_iter(["pid", "name", "cpu_percent", "memory_percent"]):
        try:
            info = p.info
            procs.append(info)
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            pass
    
    procs.sort(key=lambda x: x["cpu_percent"] or 0, reverse=True)
    
    lines = ["PID    名称                CPU%   内存%"]
    lines.append("-" * 45)
    for p in procs[:10]:
        lines.append(f"{p['pid']:<8}{p['name']:<20}{p['cpu_percent']:.1f}%    {p['memory_percent']:.1f}%")
    
    return "\n".join(lines)

4.5 定义Prompts（提示模板）

Prompts本质上是预设的上下文指令，让AI按固定格式处理特定任务：

# prompts/code_review.py
from mcp.server.fastmcp import prompt

@prompt()
def code_review(code: str, language: str = "python") -> str:
    """代码审查提示模板
    Args:
        code: 需要审查的代码
        language: 编程语言
    """
    return f"""请对以下 {language} 代码进行专业审查:

```{language}
{code}
```

请从以下维度进行审查:
1. 代码质量 — 可读性、命名规范、代码风格
2. 潜在Bug — 逻辑错误、边界条件、空指针风险
3. 安全漏洞 — 注入攻击、数据泄露、权限问题
4. 性能优化 — 时间复杂度、内存使用、I/O效率
5. 最佳实践 — 设计模式、SOLID原则、错误处理

请为每个维度给出1-5分评分和具体改进建议。"""

@prompt()
def explain_code(code: str) -> str:
    """代码解释提示模板
    Args:
        code: 需要解释的代码
    """
    return f"""请用通俗易懂的语言解释以下代码的功能:

```{code}
```

要求:
1. 先用一句话概括整体功能
2. 逐段解释关键逻辑
3. 说明输入和输出
4. 指出可能的使用场景"""

4.6 组装完整Server

将上述分散的模块全部注册到主Server中：

# server.py
from mcp.server.fastmcp import FastMCP

mcp = FastMCP(
    name="enterprise-toolkit",
    version="1.0.0",
    description="企业级MCP工具服务：天气查询、数据库操作、文件管理",
)

# ── 注册Tools ──────────────────────────────────────
from tools.weather import get_weather, get_forecast
from tools.database import query_database, list_tables
from tools.file_manager import read_file, write_file, list_files

mcp.tool(get_weather)
mcp.tool(get_forecast)
mcp.tool(query_database)
mcp.tool(list_tables)
mcp.tool(read_file)
mcp.tool(write_file)
mcp.tool(list_files)

# ── 注册Resources ──────────────────────────────────
from resources.system_info import get_system_info, get_top_processes
mcp.resource("system://info")(get_system_info)
mcp.resource("system://processes")(get_top_processes)

# ── 注册Prompts ────────────────────────────────────
from prompts.code_review import code_review, explain_code
mcp.prompt(code_review)
mcp.prompt(explain_code)

# ── 启动 ────────────────────────────────────────────
if __name__ == "__main__":
    mcp.run()

五、配置与使用

5.1 在Claude Desktop中配置

编辑Claude Desktop配置文件，路径因操作系统而异：

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

{
  "mcpServers": {
    "enterprise-toolkit": {
      "command": "python",
      "args": ["C:/path/to/my-mcp-server/server.py"],
      "env": {
        "OPENWEATHER_API_KEY": "your-api-key"
      }
    }
  }
}

5.2 在Cursor / VS Code中配置

项目根目录创建 .cursor/mcp.json：

{
  "mcpServers": {
    "enterprise-toolkit": {
      "command": "python",
      "args": ["./my-mcp-server/server.py"]
    }
  }
}

5.3 调试流程

调试链路直观流畅：启动MCP Inspector → 浏览器打开调试界面 → 浏览已注册的各类工具 → 填入参数执行测试 → 确认结果后正式配置到AI客户端。可视化调试是MCP的突出优势，大幅减少命令行反复测试的繁琐操作。

六、进阶：SSE模式部署为远程服务

6.1 使用SSE传输

# server_sse.py
from mcp.server.fastmcp import FastMCP

mcp = FastMCP(
    name="enterprise-toolkit-remote",
    host="0.0.0.0",
    port=8080,
)

# ... 注册tools/resources/prompts（同上）...

if __name__ == "__main__":
    # SSE模式：通过HTTP暴露服务
    mcp.run(transport="sse")

6.2 Docker部署

# Dockerfile
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8080
CMD ["python", "server_sse.py"]

# docker-compose.yml
version: "3.8"
services:
  mcp-server:
    build: .
    ports:
      - "8080:8080"
    environment:
      - OPENWEATHER_API_KEY=${OPENWEATHER_API_KEY}
    volumes:
      - ./data:/app/data
    restart: unless-stopped

6.3 客户端配置（远程SSE）

{
  "mcpServers": {
    "enterprise-toolkit": {
      "url": "http://your-server:8080/sse"
    }
  }
}

七、MCP协议消息流详解

当用户在Claude中说出“帮我查一下北京天气”，背后的完整消息流如下：用户输入 → Claude分析意图，决定调用get_weather工具 → 通过MCP发送JSON-RPC请求（method为tools/call，参数包含工具名和参数）→ MCP Server执行实际函数 → 返回结果（含天气数据）→ Claude结合结果生成自然语言回答 → 用户看到预期输出。尽管步骤较多，但由于内部采用高效的序列化协议，实际延迟极低。

八、最佳实践与安全注意事项

8.1 工具设计原则

设计MCP工具时，需牢记以下原则：单一职责——每个工具只完成一项功能；docstring是AI理解工具的关键，越清晰AI使用越准确；参数必须包含明确的类型标注和默认值；访问文件或数据库时必须实施沙箱限制；错误信息应友好可读，而非直接抛出异常；相同输入产生相同输出，保持幂等性。

8.2 安全清单

安全始终优先。SQL注入防护是基础——严格限定仅允许SELECT操作。路径穿越攻击需要通过resolve()校验前缀来防御。用户输入中的危险字符必须过滤。这些措施虽增加少量代码，但在生产环境中缺一不可。

# 安全示例：参数验证与路径限制
import re
from pathlib import Path

def validate_sql(sql: str) -> bool:
    """只允许SELECT查询"""
    forbidden = ["DROP", "DELETE", "INSERT", "UPDATE", "ALTER", "CREATE"]
    upper_sql = sql.upper()
    return all(kw not in upper_sql for kw in forbidden)

def safe_filepath(base_dir: Path, user_path: str) -> Path:
    """防止路径穿越攻击"""
    full_path = (base_dir / user_path).resolve()
    if not str(full_path).startswith(str(base_dir.resolve())):
        raise ValueError("非法路径访问")
    return full_path

def sanitize_input(text: str) -> str:
    """清理用户输入"""
    # 移除潜在的危险字符
    return re.sub(r'[<>&\']', '', text)

九、总结

从零到一，我们完整走了一遍MCP Server的开发流程。关键知识点回顾：FastMCP是Python生态中最核心的开发框架；@tool()、@resource()、@prompt()三个装饰器分别对应不同能力；stdio适合本地开发，SSE适合远程部署；MCP Inspector是调试利器，无需命令行手动测试。下一阶段可探索的方向包括：将MCP Server接入企业内部API、开发多用户权限管理中间件、结合RAG构建知识库服务，甚至参与开源生态发布自有工具。

MCP协议正快速成为AI工具调用的行业标准，掌握它意味着能让AI真正融入你的工作流。这不是锦上添花，而是从“对话式AI”到“行动式AI”的跨越。

来源：互联网