使用Python开发你的第一个AI工具服务MCP Server

┌──────────────────────────────────────────────────────────────┐
│                     MCP 架构全景                              │
│                                                              │
│  ┌──────────┐    ┌──────────┐    ┌──────────────────────┐    │
│  │  AI 应用  │    │ MCP 客户端 │    │    MCP 服务端 (Server) │    │
│  │  (Host)  │◄──►│ (Client)  │◄──►│                      │    │
│  │          │    │           │    │  ┌────────┐          │    │
│  │ Claude   │    │ 内置于    │    │  │ Tools  │ 工具调用  │    │
│  │ Cursor   │    │ Host 应用 │    │  │        │          │    │
│  │ VS Code  │    │ 之中      │    │  ├────────┤          │    │
│  │          │    │           │    │  │Resources│ 数据读取  │    │
│  └──────────┘    └──────────┘    │  │        │          │    │
│                                  │  ├────────┤          │    │
│                                  │  │Prompts │ 提示模板  │    │
│                                  │  └────────┘          │    │
│                                  └──────────────────────┘    │
└──────────────────────────────────────────────────────────────┘

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

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

pip install "mcp[cli]" httpx

mcp --version

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

# 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")
# 启动 Server
if __name__ == "__main__":
    mcp.run()

# 使用 MCP Inspector 测试（可视化调试工具）
mcp dev server.py
# 或直接运行
python server.py

# 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)

# 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 "查询结果为空"

        # 格式化为 Markdown 表格
        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\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)

# 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 "目录为空"

# 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

    # 按 CPU 使用率排序
    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)

# 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]

Args:
    code: 需要解释的代码
"""
return f"""请用通俗易懂的语言解释以下代码的功能:

[code]

### 4.6 组装完整 Server

```python
# 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()

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

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

┌─────────────────────────────────────────────────────────┐
│                    MCP 调试流程                          │
│                                                         │
│  1. 启动 Inspector                                     │
│     $ mcp dev server.py                                │
│                     │                                   │
│                     ▼                                   │
│  2. 浏览器打开 http://localhost:5173                     │
│                     │                                   │
│                     ▼                                   │
│  3. 查看已注册的 Tools / Resources / Prompts            │
│     ┌─────────────────────────────────────┐            │
│     │ Tools:                              │            │
│     │   ✓ get_weather                     │            │
│     │   ✓ get_forecast                    │            │
│     │   ✓ query_database                  │            │
│     │   ✓ read_file                       │            │
│     │   ✓ write_file                      │            │
│     │   ✓ list_files                      │            │
│     │                                     │            │
│     │ Resources:                          │            │
│     │   ✓ system://info                   │            │
│     │   ✓ system://processes              │            │
│     │                                     │            │
│     │ Prompts:                            │            │
│     │   ✓ code_review                     │            │
│     │   ✓ explain_code                    │            │
│     └─────────────────────────────────────┘            │
│                     │                                   │
│                     ▼                                   │
│  4. 点击工具 → 填入参数 → 点击 Run → 查看返回结果       │
│                     │                                   │
│                     ▼                                   │
│  5. 确认无误后，配置到 AI 客户端正式使用                  │
└─────────────────────────────────────────────────────────┘

# 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")

# 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

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

用户: "帮我查一下北京天气"
  │
  ▼
┌──────────┐
│   Claude  │ 1. 分析意图，决定调用 tool: get_weather
│  (Host)  │
└────┬─────┘
     │ MCP Request (JSON-RPC)
     │ {
     │   "jsonrpc": "2.0",
     │   "method": "tools/call",
     │   "params": {
     │     "name": "get_weather",
     │     "arguments": {
     │       "city": "北京"
     │     }
     │   }
     │ }
     ▼
┌──────────┐
│MCP Server│ 2. 执行 get_weather("北京")
│ (Python) │
└────┬─────┘
     │ MCP Response
     │ {
     │   "jsonrpc": "2.0",
     │   "result": {
     │     "content": [{
     │       "type": "text",
     │       "text": "【北京 天气】\n天气: 晴\n温度: 22°C..."
     │     }]
     │   }
     │ }
     ▼
┌──────────┐
│   Claude  │ 3. 结合工具返回结果，生成自然语言回答
│  (Host)  │
└────┬─────┘
     │
     ▼
用户看到: "北京今天天气晴朗，气温 22°C，湿度 35%，风速 3 m/s。"

┌───────────────────────────────────────────────────┐
│              MCP 工具设计原则                       │
├───────────────────────────────────────────────────┤
│                                                   │
│  1. 单一职责 ─── 每个工具只做一件事                 │
│                                                   │
│  2. 清晰描述 ─── docstring 是 AI 理解工具的关键     │
│                                                   │
│  3. 类型标注 ─── 参数必须有明确的类型和默认值        │
│                                                   │
│  4. 安全沙箱 ─── 限制文件/数据库访问范围             │
│                                                   │
│  5. 错误友好 ─── 返回清晰的错误信息，而非抛异常      │
│                                                   │
│  6. 幂等设计 ─── 相同输入应产生相同输出              │
│                                                   │
└───────────────────────────────────────────────────┘

# 安全示例：参数验证与路径限制
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)