MCP协议实战：Python开发首个AI工具服务

大型语言模型的能力正在被赋予“双手”——借助MCP协议，AI应用如今能够真正执行代码、调用API、操作文件，而不仅仅是生成文本。本文将从零开始，完整呈现一个MCP Server的实战开发全流程，帮助你快速上手这一关键技术。

一、MCP协议是什么？

1.1 核心概念

MCP（Model Context Protocol）定义了一套标准化协议，让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进行网络通信。两种方式各有优势：本地开发追求便捷，生产环境则必须依赖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"