手把手教你编写最简 MCP 服务器（附完整代码）

核心主题与目的

• 演示如何用 Python 编写一个最基础的 MCP（Model Context Protocol）服务，并配置 AI 客户端（Claude Desktop）调用该服务。
• 说明 MCP 的作用：将 AI Agent 与工具解耦，让 AI 模型根据用户请求自主决定调用哪个工具。

重要概念与术语

• MCP：Anthropic 提出的工具托管协议，Agent 从 MCP 服务收集可用工具信息，与用户请求一起发给 AI 模型，模型返回指令引导 Agent 调用本地工具。
• AI Agent：负责执行 AI 模型指令的程序，本身不做判断，只执行调用。
• 工具函数：被 MCP 暴露给 AI 的普通 Python 函数，返回值需可序列化为字典。
• 通信模式：
  • STDIO：标准输入输出，简单，只能与 Agent 同机运行。
  • SSE：基于 HTTP，可远程部署，但需处理认证授权。

实现MCP服务

• MCP 服务的核心实现步骤
  • 创建 MCP 实例：FastMCP("服务名")，名称建议体现功能。
  • 注册工具：用 add_tool(函数) 或装饰器 @mcp.tool() 将函数暴露给 AI。
  • 启动服务：mcp.run(transport="stdio") 选择通信模式。
• 工具函数的编写注意事项
  • 函数名、参数名清晰，加上类型标注。
  • 复杂功能需写 Docstring，AI 模型会读取这些信息，就像程序员阅读 API 文档。
  • 返回值格式不限，能转成字典即可被 AI 理解。

基础MCP服务

import platform
import psutil
import subprocess
import json
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("host info mcp")


def get_host_info() -> str:
    """get host information
    Returns:
        str: the host information in JSON string
    """
    info: dict[str, str] = {
        "system": platform.system(),
        "release": platform.release(),
        "machine": platform.machine(),
        "processor": platform.processor(),
        "memory_gb": str(round(psutil.virtual_memory().total / (1024 ** 3), 2)),
    }

    cpu_count = psutil.cpu_count(logical=True)
    if cpu_count is None:
        info["cpu_count"] = "-1"
    else:
        info["cpu_count"] = str(cpu_count)

    try:
        cpu_model = subprocess.check_output(
            ["sysctl", "-n", "machdep.cpu.brand_string"]
        ).decode().strip()
        info["cpu_model"] = cpu_model
    except Exception:
        info["cpu_model"] = "Unknown"

    return json.dumps(info)


@mcp.tool()
def user_info():
    """get user information"""
    # 装饰器注册与 add_tool() 方法注册完全等价，可按代码组织方式任选。
    return "用户叫小明，当前24岁"


def start_mcp_server():
    """
    当前业界绝大多数 MCP 服务采用 stdio 模式，因其部署简单；
    SSE 适用于需要跨网络/云原生部署的场景。
    真正困难的不是写 MCP 服务本身，而是设计出边界清晰、语义明确、恰好解决真实问题的工具。
    """
    mcp.add_tool(get_host_info)
    mcp.run("stdio")


if __name__ == "__main__":
    # print(get_host_info())
    # print(user_info())
    start_mcp_server()

Agent 中使用 MCP 服务

from fastmcp.client import StdioTransport
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPToolset
from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.providers.deepseek import DeepSeekProvider

from config import DEEPSEEK_API_KEY

mcp_toolset = MCPToolset(StdioTransport(
    command=r"D:\jaifiresong\bili2text\.venv\Scripts\python.exe",
    args=[r"D:\gitee\ai_app_dev\02\1.py"]
))

model = OpenAIChatModel(
    "deepseek-chat",
    provider=DeepSeekProvider(api_key=DEEPSEEK_API_KEY),
)

agent = Agent(
    model,
    system_prompt="你是一个有用的AI助手，可以查询主机信息和用户信息。",
    toolsets=[mcp_toolset],
)


def main():
    history = []
    while True:
        user_input = input("Input: ")
        if user_input.lower() in {"exit", "quit"}:
            break
        resp = agent.run_sync(user_input, message_history=history)
        history = list(resp.all_messages())
        print(resp.output)


if __name__ == "__main__":
    main()

手动测试MCP服务

import json
import subprocess
import sys
import time

"""
MCP 服务测试脚本
通过 stdio 与 MCP 服务进行 JSON-RPC 通信，验证服务是否正常工作。

脚本功能：
- 通过 stdio 管道进行 JSON-RPC 通信，依次验证：
   - initialize 初始化握手
       - initialize 请求是 MCP 握手流程的第一步，它的核心作用是建立客户端与服务器之间的会话（Session）基础。如果缺少这一步或握手失败，后续所有的工具调用、资源读取等操作都会被服务器拒绝。
   - notifications/initialized 通知
       - 它的核心作用是 告诉服务器：“我已收到你的初始化响应，一切正常，我们现在可以正式开始干活了。”
       - 不发虽然能跑，但强烈建议发送它。这只是一个空 JSON 包（几乎没有性能开销），却能保证你的代码在任何遵循标准的 MCP 服务器上都能稳定运行。
   - tools/list 工具列表（检查 get_host_info 和 foo 是否存在）
   - tools/call 调用 get_host_info 工具并验证返回 JSON 结构
- 自动清理子进程
"""


def send_message(_proc, msg: dict) -> None:
    """向 MCP 服务发送一条 JSON-RPC 消息（追加换行符）。"""
    line = json.dumps(msg, ensure_ascii=False)
    _proc.stdin.write(line.encode("utf-8") + b"\n")
    _proc.stdin.flush()
    print(f"[C -> S] {line}")


def read_message(_proc, timeout: float = 10.0) -> dict | None:
    """从 MCP 服务读取一条 JSON-RPC 消息。"""
    try:
        line = _proc.stdout.readline()
        if not line:
            return None
        decoded = line.decode("utf-8").strip()
        print(f"[S -> C] {decoded}")
        return json.loads(decoded)
    except Exception as e:
        print(f"[ERROR] 读取消息失败: {e}")
        return None


proc: subprocess.Popen | None = None


def start_mcp_server() -> subprocess.Popen:
    """运行 MCP 服务测试。"""
    print("=" * 60)
    print("启动 MCP 服务测试...")
    print("=" * 60)

    # 1. 启动 MCP 服务子进程
    python_cmd = [
        r"D:\jaifiresong\bili2text\.venv\Scripts\python.exe",  # 使用指定的 Python 虚拟环境解释器
        r"1.py",
    ]

    global proc
    proc = subprocess.Popen(
        python_cmd,
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )

    # 等待服务启动
    time.sleep(0.5)

    # 检查子进程是否已异常退出
    if proc.poll() is not None:
        stderr_data = proc.stderr.read()
        print("[FAIL] MCP 服务启动失败，进程已退出")
        if stderr_data:
            print("[STDERR]")
            print(stderr_data.decode("utf-8", errors="replace"))
        raise Exception("MCP 服务启动失败")

    return proc


# 2. 发送 initialize 请求
def t1() -> None:
    print("\n[步骤 1] 发送 initialize 请求...")
    init_id = "init-1"
    send_message(
        proc,
        {
            "jsonrpc": "2.0",
            "id": init_id,
            "method": "initialize",
            "params": {
                "protocolVersion": "2024-11-05",
                "capabilities": {},
                "clientInfo": {"name": "test-client", "version": "1.0.0"},
            },
        },
    )

    response = read_message(proc)
    if response is None:
        print("[FAIL] 未收到 initialize 响应")
    if response.get("id") != init_id:
        print("[FAIL] initialize 响应 ID 不匹配")
    if "result" not in response:
        print("[FAIL] initialize 响应中没有 result")
    print("[PASS] initialize 成功")


def t2() -> None:
    print("\n[步骤 2] 发送 initialized 通知...")
    send_message(
        proc,
        {
            "jsonrpc": "2.0",
            "method": "notifications/initialized",
        },
    )
    print("[PASS] initialized 通知已发送")


def t3() -> None:
    print("\n[步骤 3] 请求工具列表 (tools/list)...")
    list_id = "list-1"
    send_message(
        proc,
        {
            "jsonrpc": "2.0",
            "id": list_id,
            "method": "tools/list",
            "params": {},
        },
    )

    response = read_message(proc)
    if response is None:
        print("[FAIL] 未收到 tools/list 响应")
        return False
    if response.get("id") != list_id:
        print("[FAIL] tools/list 响应 ID 不匹配")
        return False

    tools_result = response.get("result", {})
    tools = tools_result.get("tools", [])
    tool_names = [t.get("name") for t in tools]
    print(f"[INFO] 发现工具: {tool_names}")


def t4() -> None:
    print("\n[步骤 5] 调用 get_host_info 工具...")
    call_id2 = "call-2"
    send_message(
        proc,
        {
            "jsonrpc": "2.0",
            "id": call_id2,
            "method": "tools/call",
            "params": {"name": "get_host_info", "arguments": {}},
        },
    )

    response = read_message(proc)
    if response is None:
        print("[FAIL] 未收到 get_host_info 调用响应")
        return False
    if response.get("id") != call_id2:
        print("[FAIL] get_host_info 调用响应 ID 不匹配")
        return False

    content = response.get("result", {}).get("content", [])
    text_parts = [c.get("text", "") for c in content if c.get("type") == "text"]
    full_text = "".join(text_parts)
    try:
        host_info = json.loads(full_text)
        print(f"[PASS] get_host_info 工具调用成功 返回：{host_info}")
    except json.JSONDecodeError:
        print(f"[FAIL] get_host_info 返回的不是合法 JSON: {full_text}")
        return False


if __name__ == "__main__":
    try:
        start_mcp_server()
        t1()
        # t2()
        # t3()
        t4()
    finally:
        # 清理子进程
        print("\n[清理] 终止 MCP 服务子进程...")
        try:
            proc.stdin.close()
        except Exception:
            pass
        proc.terminate()
        try:
            proc.wait(timeout=2)
        except subprocess.TimeoutExpired:
            proc.kill()
            proc.wait()

思维框架

• 将 AI 模型视为程序员：工具函数就是提供给程序员的 API，函数签名、类型标注和 Docstring 就是 API 文档，清晰程度决定 AI 能否正确使用。
• 工具设计的本质：提供工具不仅是暴露函数，更是传递你对问题的理解和观察世界的方式。

结论或行动启示

• 实现一个 MCP 服务代码量很少，真正的难点在于设计出有意义、足够清晰、能准确解决问题的工具。
• 想让 AI 理解我们，首先要自己看清问题是什么，将这种理解转化为工具的设计。