用 Python 构建 MCP 服务器：完整分步教程

# macOS / Linux
curl -LsSf https://astral.sh/uv/install.sh | sh

# 或者用 Homebrew
brew install uv

# 验证安装
uv --version

# 用 uv 创建项目
uv init weather-mcp-server
cd weather-mcp-server

# 添加带 CLI 工具的 MCP SDK
uv add "mcp[cli]"

mkdir weather-mcp-server
cd weather-mcp-server
python3 -m venv .venv
source .venv/bin/activate
pip install "mcp[cli]"

[project]
name = "weather-mcp-server"
version = "1.0.0"
description = "MCP server providing weather data to AI assistants"
requires-python = ">=3.10"
dependencies = [
    "mcp[cli]>=1.2.0",
]

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project.scripts]
weather-mcp-server = "weather_mcp_server.server:main"

mkdir -p src/weather_mcp_server
touch src/weather_mcp_server/__init__.py
touch src/weather_mcp_server/server.py

weather-mcp-server/
├── pyproject.toml
├── src/
│   └── weather_mcp_server/
│       ├── __init__.py
│       └── server.py       # 主服务器代码
└── README.md

"""天气 MCP 服务器 - 为 AI 助手提供天气数据。"""

from mcp.server.fastmcp import FastMCP

# 创建 FastMCP 服务器实例
mcp = FastMCP("weather-server")

# 模拟天气数据（在生产环境中替换为真实 API）
WEATHER_DATA = {
    "New York": {"temp_c": 2, "condition": "Cloudy", "humidity": 65, "wind_kmh": 20},
    "San Francisco": {"temp_c": 14, "condition": "Sunny", "humidity": 70, "wind_kmh": 15},
    "London": {"temp_c": 8, "condition": "Rainy", "humidity": 85, "wind_kmh": 25},
    "Tokyo": {"temp_c": 10, "condition": "Clear", "humidity": 55, "wind_kmh": 10},
    "Sydney": {"temp_c": 25, "condition": "Sunny", "humidity": 60, "wind_kmh": 18},
    "Berlin": {"temp_c": 5, "condition": "Overcast", "humidity": 75, "wind_kmh": 22},
}


@mcp.tool()
def get_weather(city: str) -> str:
    """获取指定城市的当前天气。

    Args:
        city: 城市名称，例如 "New York"、"London"、"Tokyo"

    Returns:
        天气报告，包含温度、天气状况、湿度和风速。
    """
    data = WEATHER_DATA.get(city)
    if not data:
        return f"没有 '{city}' 的天气数据。可用城市：{', '.join(WEATHER_DATA.keys())}"

    return (
        f"{city} 的天气：\n"
        f"  温度：{data['temp_c']}°C\n"
        f"  天气：{data['condition']}\n"
        f"  湿度：{data['humidity']}%\n"
        f"  风速：{data['wind_kmh']} km/h"
    )


@mcp.tool()
def compare_weather(city1: str, city2: str) -> str:
    """比较两个城市的天气。

    Args:
        city1: 第一个城市名称
        city2: 第二个城市名称

    Returns:
        两个城市的天气并排对比，包含温差。
    """
    data1 = WEATHER_DATA.get(city1)
    data2 = WEATHER_DATA.get(city2)

    if not data1:
        return f"没有 '{city1}' 的数据。可用城市：{', '.join(WEATHER_DATA.keys())}"
    if not data2:
        return f"没有 '{city2}' 的数据。可用城市：{', '.join(WEATHER_DATA.keys())}"

    diff = data1["temp_c"] - data2["temp_c"]
    warmer = city1 if diff > 0 else city2

    return (
        f"天气对比：{city1} vs {city2}\n"
        f"{'─' * 40}\n"
        f"{city1}：{data1['condition']}，{data1['temp_c']}°C，"
        f"湿度 {data1['humidity']}%，风速 {data1['wind_kmh']} km/h\n"
        f"{city2}：{data2['condition']}，{data2['temp_c']}°C，"
        f"湿度 {data2['humidity']}%，风速 {data2['wind_kmh']} km/h\n"
        f"{'─' * 40}\n"
        f"温差：{abs(diff)}°C（{warmer} 更暖）"
    )


def main():
    """服务器入口点。"""
    mcp.run()


if __name__ == "__main__":
    main()

@mcp.resource("weather://cities")
def list_cities() -> str:
    """列出所有有天气数据的城市。

    返回支持的城市名称的 JSON 数组。
    """
    import json
    return json.dumps(list(WEATHER_DATA.keys()), indent=2)


@mcp.resource("weather://city/{city_name}")
def get_city_details(city_name: str) -> str:
    """以 JSON 格式获取指定城市的详细天气数据。

    此资源模板接受 URL 中的任意城市名称。
    """
    import json
    data = WEATHER_DATA.get(city_name)
    if not data:
        return json.dumps({"error": f"未找到城市 '{city_name}'"})
    return json.dumps({"city": city_name, **data}, indent=2)

@mcp.prompt()
def travel_advisory(destination: str) -> str:
    """根据目的地的当前天气获取旅行建议。

    Args:
        destination: 你计划前往的城市
    """
    data = WEATHER_DATA.get(destination)
    if not data:
        return f"请查看 {destination} 的天气状况，并建议合适的穿着和旅行贴士。"

    return (
        f"{destination} 当前天气为 {data['condition']}，"
        f"温度 {data['temp_c']}°C，湿度 {data['humidity']}%，"
        f"风速 {data['wind_kmh']} km/h。\n\n"
        f"根据这些条件，请建议：\n"
        f"1. 适合携带的衣物\n"
        f"2. 天气相关的旅行贴士\n"
        f"3. 户外活动的最佳时段"
    )


@mcp.prompt()
def weather_report(cities: str = "all") -> str:
    """生成全面的天气报告。

    Args:
        cities: 逗号分隔的城市名称，或 "all" 表示所有城市
    """
    if cities == "all":
        city_list = list(WEATHER_DATA.keys())
    else:
        city_list = [c.strip() for c in cities.split(",")]

    city_str = ", ".join(city_list)
    return (
        f"请为以下城市生成专业的天气报告：{city_str}。\n\n"
        f"包含：\n"
        f"- 当前天气概况\n"
        f"- 温度趋势\n"
        f"- 旅行建议\n"
        f"- 天气预警（如有）"
    )

import httpx

@mcp.tool()
async def fetch_weather_api(city: str) -> str:
    """从外部 API 获取实时天气数据。

    Args:
        city: 要查询的城市名称
    """
    try:
        async with httpx.AsyncClient() as client:
            response = await client.get(
                "https://api.weatherapi.com/v1/current.json",
                params={"key": "YOUR_API_KEY", "q": city},
                timeout=10.0,
            )
            response.raise_for_status()
            data = response.json()

            current = data["current"]
            return (
                f"{city} 的天气：\n"
                f"  温度：{current['temp_c']}°C\n"
                f"  天气：{current['condition']['text']}\n"
                f"  湿度：{current['humidity']}%\n"
                f"  风速：{current['wind_kph']} km/h"
            )

    except httpx.TimeoutException:
        return f"错误：获取 '{city}' 天气时请求超时。请稍后重试。"
    except httpx.HTTPStatusError as e:
        return f"错误：API 为 '{city}' 返回了状态码 {e.response.status_code}。"
    except Exception as e:
        return f"错误：无法获取 '{city}' 的天气：{str(e)}"

from mcp.server.fastmcp import Context

@mcp.tool()
async def analyze_weather_trends(city: str, ctx: Context) -> str:
    """分析城市的天气趋势，带进度报告。

    Args:
        city: 要分析的城市名称
    """
    # 日志消息（发送到 MCP 客户端）
    await ctx.info(f"开始分析 {city} 的天气")

    # 报告进度（对长时间运行的操作很有用）
    await ctx.report_progress(0, 3)

    await ctx.debug("正在获取当前数据...")
    data = WEATHER_DATA.get(city)
    if not data:
        await ctx.warning(f"未找到 {city} 的数据")
        return f"没有 '{city}' 的可用数据"

    await ctx.report_progress(1, 3)

    await ctx.debug("正在计算分析结果...")
    # 模拟分析工作
    analysis = f"{city} 的天气分析：{data['condition']} 天气，{data['temp_c']}°C"

    await ctx.report_progress(2, 3)

    await ctx.info(f"{city} 的分析完成")
    await ctx.report_progress(3, 3)

    return analysis

import sys

# 错误方式 - 破坏 STDIO 协议
print("Debug: processing request")

# 正确方式 - 使用 stderr，不会干扰协议
print("Debug: processing request", file=sys.stderr)

# 启动 Inspector
mcp dev src/weather_mcp_server/server.py

# 或者用 uv（如果不在激活的 venv 中）
uv run mcp dev src/weather_mcp_server/server.py

# 以 stdio 传输方式运行（默认）
mcp run src/weather_mcp_server/server.py

# 使用 uv（推荐 - 自动处理虚拟环境）
claude mcp add weather-server -- uv run --directory /absolute/path/to/weather-mcp-server mcp run src/weather_mcp_server/server.py

# 或者使用 pyproject.toml 中定义的脚本入口点
claude mcp add weather-server -- uv run --directory /absolute/path/to/weather-mcp-server weather-mcp-server

claude mcp add weather-server /absolute/path/to/weather-mcp-server/.venv/bin/python -m weather_mcp_server.server

# 检查服务器是否已注册
claude mcp list

> 东京的天气怎么样？

> 比较一下纽约和伦敦的天气

> 列出天气服务器中所有可用的城市

claude mcp remove weather-server
claude mcp add weather-server -- uv run --directory /absolute/path/to/weather-mcp-server mcp run src/weather_mcp_server/server.py

"""SQLite MCP 服务器 - 让 AI 助手以只读方式访问 SQLite 数据库。"""

import sqlite3
import json
from pathlib import Path
from contextlib import closing

from mcp.server.fastmcp import FastMCP, Context

# 创建服务器
mcp = FastMCP("sqlite-server")

# 数据库路径 - 根据你的项目配置
DB_PATH = Path("./data/example.db")


def get_connection() -> sqlite3.Connection:
    """创建带安全设置的数据库连接。"""
    conn = sqlite3.connect(str(DB_PATH))
    conn.row_factory = sqlite3.Row  # 以字典形式返回行
    conn.execute("PRAGMA journal_mode=WAL")  # 更好的并发读取性能
    return conn


def init_sample_db():
    """创建示例数据库用于演示。"""
    DB_PATH.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(str(DB_PATH))
    cursor = conn.cursor()

    cursor.executescript("""
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL,
            email TEXT UNIQUE NOT NULL,
            role TEXT DEFAULT 'user',
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        );

        CREATE TABLE IF NOT EXISTS projects (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL,
            description TEXT,
            owner_id INTEGER REFERENCES users(id),
            status TEXT DEFAULT 'active',
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        );

        CREATE TABLE IF NOT EXISTS tasks (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            title TEXT NOT NULL,
            project_id INTEGER REFERENCES projects(id),
            assignee_id INTEGER REFERENCES users(id),
            status TEXT DEFAULT 'todo',
            priority TEXT DEFAULT 'medium',
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        );

        -- 如果表为空则插入示例数据
        INSERT OR IGNORE INTO users (id, name, email, role) VALUES
            (1, 'Alice Chen', '[email protected]', 'admin'),
            (2, 'Bob Smith', '[email protected]', 'developer'),
            (3, 'Carol Wu', '[email protected]', 'developer'),
            (4, 'Dave Brown', '[email protected]', 'designer');

        INSERT OR IGNORE INTO projects (id, name, description, owner_id, status) VALUES
            (1, 'MCP Server', 'Build custom MCP servers', 1, 'active'),
            (2, 'Web Dashboard', 'Admin dashboard redesign', 4, 'active'),
            (3, 'API Migration', 'Migrate REST to GraphQL', 2, 'completed');

        INSERT OR IGNORE INTO tasks (id, title, project_id, assignee_id, status, priority) VALUES
            (1, 'Set up project structure', 1, 1, 'done', 'high'),
            (2, 'Implement weather tool', 1, 2, 'in_progress', 'high'),
            (3, 'Write unit tests', 1, 3, 'todo', 'medium'),
            (4, 'Design dashboard layout', 2, 4, 'in_progress', 'high'),
            (5, 'Implement auth flow', 2, 2, 'todo', 'high'),
            (6, 'Write API docs', 3, 3, 'done', 'low'),
            (7, 'Performance optimization', 1, 2, 'todo', 'medium');
    """)

    conn.commit()
    conn.close()


@mcp.tool()
def query_database(sql: str) -> str:
    """对数据库执行只读 SQL 查询。

    仅允许 SELECT 语句。查询在只读事务中运行以防止任何数据修改。

    Args:
        sql: 要执行的 SELECT SQL 查询

    Returns:
        格式化的 JSON 字符串形式的查询结果，或错误消息。
    """
    # 安全性：仅允许 SELECT 语句
    normalized = sql.strip().upper()
    if not normalized.startswith("SELECT"):
        return "错误：仅允许 SELECT 查询。此服务器提供只读访问。"

    # 阻止危险关键字（包括子查询中的）
    dangerous_keywords = ["INSERT", "UPDATE", "DELETE", "DROP", "ALTER", "CREATE", "EXEC"]
    for keyword in dangerous_keywords:
        if keyword in normalized:
            return f"错误：不允许 '{keyword}' 语句。"

    try:
        with closing(get_connection()) as conn:
            cursor = conn.execute(sql)
            columns = [description[0] for description in cursor.description]
            rows = [dict(zip(columns, row)) for row in cursor.fetchall()]

            return json.dumps({
                "columns": columns,
                "row_count": len(rows),
                "rows": rows,
            }, indent=2, default=str)

    except sqlite3.OperationalError as e:
        return f"SQL 错误：{str(e)}"
    except Exception as e:
        return f"执行查询时出错：{str(e)}"


@mcp.tool()
def list_tables() -> str:
    """列出数据库中所有表及其列定义。

    Returns:
        每个表及其列的格式化描述。
    """
    try:
        with closing(get_connection()) as conn:
            # 获取所有表名
            cursor = conn.execute(
                "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
            )
            tables = [row[0] for row in cursor.fetchall()]

            result = []
            for table in tables:
                # 获取每个表的列信息
                cursor = conn.execute(f"PRAGMA table_info({table})")
                columns = []
                for col in cursor.fetchall():
                    col_name = col[1]
                    col_type = col[2]
                    not_null = "NOT NULL" if col[3] else ""
                    pk = "PRIMARY KEY" if col[5] else ""
                    columns.append(f"    {col_name} {col_type} {pk} {not_null}".strip())

                # 获取行数
                cursor = conn.execute(f"SELECT COUNT(*) FROM {table}")
                row_count = cursor.fetchone()[0]

                result.append(
                    f"表：{table}（{row_count} 行）\n" +
                    "\n".join(columns)
                )

            return "\n\n".join(result)

    except Exception as e:
        return f"列出表时出错：{str(e)}"


@mcp.tool()
async def describe_table(table_name: str, ctx: Context) -> str:
    """获取指定表的详细 schema 和示例数据。

    Args:
        table_name: 要描述的表名

    Returns:
        表 schema、示例行和基本统计信息。
    """
    await ctx.info(f"正在描述表：{table_name}")

    try:
        with closing(get_connection()) as conn:
            # 验证表是否存在
            cursor = conn.execute(
                "SELECT name FROM sqlite_master WHERE type='table' AND name=?",
                (table_name,)
            )
            if not cursor.fetchone():
                return f"错误：表 '{table_name}' 不存在。"

            # 获取 schema
            cursor = conn.execute(f"PRAGMA table_info({table_name})")
            columns = cursor.fetchall()

            schema_lines = [f"'{table_name}' 的 Schema："]
            for col in columns:
                schema_lines.append(
                    f"  {col[1]} ({col[2]})"
                    + (" PRIMARY KEY" if col[5] else "")
                    + (" NOT NULL" if col[3] else "")
                    + (f" DEFAULT {col[4]}" if col[4] else "")
                )

            # 获取行数
            cursor = conn.execute(f"SELECT COUNT(*) FROM {table_name}")
            count = cursor.fetchone()[0]

            # 获取示例行（前 5 行）
            cursor = conn.execute(f"SELECT * FROM {table_name} LIMIT 5")
            col_names = [desc[0] for desc in cursor.description]
            sample_rows = [dict(zip(col_names, row)) for row in cursor.fetchall()]

            result = "\n".join(schema_lines)
            result += f"\n\n总行数：{count}"
            result += f"\n\n示例数据（前 5 行）：\n{json.dumps(sample_rows, indent=2, default=str)}"

            return result

    except Exception as e:
        return f"描述表时出错：{str(e)}"


@mcp.resource("db://schema")
def get_full_schema() -> str:
    """以资源形式获取完整的数据库 schema。

    这对于需要在编写查询前了解数据库结构的 AI 助手很有用。
    """
    try:
        with closing(get_connection()) as conn:
            cursor = conn.execute(
                "SELECT sql FROM sqlite_master WHERE type='table' AND sql IS NOT NULL"
            )
            statements = [row[0] for row in cursor.fetchall()]
            return "\n\n".join(statements)
    except Exception as e:
        return f"错误：{str(e)}"


@mcp.prompt()
def data_analysis(question: str) -> str:
    """根据数据库 schema 生成数据分析提示词。

    Args:
        question: 你想回答的数据问题
    """
    schema = get_full_schema()
    return (
        f"你可以访问一个具有以下 schema 的 SQLite 数据库：\n\n"
        f"{schema}\n\n"
        f"用户想知道：{question}\n\n"
        f"请使用 query_database 工具编写并执行合适的 SQL 查询，"
        f"然后清楚地解释结果。"
    )


def main():
    """初始化示例数据库并启动服务器。"""
    init_sample_db()
    mcp.run()


if __name__ == "__main__":
    main()

# 使用 MCP Inspector 测试
mcp dev src/weather_mcp_server/db_server.py

claude mcp add sqlite-server -- uv run --directory /absolute/path/to/weather-mcp-server python src/weather_mcp_server/db_server.py

> 数据库里有哪些表？

> 显示所有活跃的项目及其负责人

> 每个开发者有多少任务？按状态分组

> 找出所有未完成的高优先级任务

# 这是调用 mcp.run() 时的默认方式
mcp.run()

# 显式指定 stdio 传输
mcp.run(transport="stdio")

mcp.run(transport="sse", host="0.0.0.0", port=8000)

claude mcp add remote-weather --url http://your-server:8000/sse

mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)

FROM python:3.12-slim

WORKDIR /app

# 安装 uv 以快速解析依赖
RUN pip install uv

# 复制项目文件
COPY pyproject.toml .
COPY src/ ./src/

# 安装依赖
RUN uv pip install --system -e .

# 以 streamable-http 传输方式运行用于远程访问
EXPOSE 8000
CMD ["python", "-m", "weather_mcp_server.server"]

import os

def main():
    transport = os.environ.get("MCP_TRANSPORT", "stdio")
    if transport == "stdio":
        mcp.run()
    else:
        mcp.run(
            transport="streamable-http",
            host="0.0.0.0",
            port=int(os.environ.get("MCP_PORT", "8000")),
        )

# 构建包
uv build

# 发布到 PyPI
uv publish

# 用户使用 uvx（uv 的工具运行器）安装并运行
claude mcp add weather-server -- uvx weather-mcp-server

# 或者用 pip
pip install weather-mcp-server
claude mcp add weather-server weather-mcp-server