Python MCP 工具开发入门：Server、Client 与 LLM 集成

核心概念解析：

• FastMCP - 简化的 MCP 服务器框架
• @mcp.tool() - 装饰器定义工具函数
• @mcp.resource() - 装饰器定义资源（带 URI 模式）
• mcp.run('stdio') - 通过标准输入输出运行服务器

3. 第二步：创建 MCP 客户端

3.1 基础客户端（无 LLM）

client.py 展示了如何直接调用工具：

import sys
import asyncio
from mcp import ClientSession
from mcp.client.stdio import stdio_client, StdioServerParameters

async def main():
    # 1. 启动 MCP 服务器进程
    server_script = "server.py"
    params = StdioServerParameters(
        command=sys.executable,
        args=[server_script],
    )
    transport = stdio_client(params)
    stdio, write = await transport.__aenter__()

    # 2. 建立客户端会话
    session = await ClientSession(stdio, write).__aenter__()
    await session.initialize()

    # 3. 调用工具
    result = await session.call_tool("add", {"a": 3, "b": 5})
    print(f"3 + 5 = {result}")

    # 4. 关闭连接
    await session.__aexit__(None, None, None)
    await transport.__aexit__(None, None, None)

if __name__ == "__main__":
    asyncio.run(main())

3.2 与 LLM 集成的客户端

client-qwen.py 展示了如何让 LLM 自动调用工具：

import sys
import asyncio
import os
import json
from mcp import ClientSession
from mcp.client.stdio import stdio_client, StdioServerParameters
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

async def main():
    # 1. 连接到 MCP 服务器
    print(">>> 初始化加法 LLM 工具客户端")
    server_script = "server.py"
    params = StdioServerParameters(
        command=sys.executable,
        args=[server_script],
    )
    transport = stdio_client(params)
    stdio, write = await transport.__aenter__()
    session = await ClientSession(stdio, write).__aenter__()
    await session.initialize()
    print(">>> 连接到 MCP 服务器成功")

    # 2. 初始化 LLM 客户端（使用通义千问）
    client = OpenAI(
        api_key=os.getenv("QWEN_API_KEY"),
        base_url=os.getenv("QWEN_BASE_URL")
    )

    # 3. 从 MCP 服务器获取工具列表
    resp = await session.list_tools()
    tools = [
        {
            "type": "function",
            "function": {
                "name": tool.name,
                "description": tool.description,
                "parameters": tool.inputSchema
            }
        }
        for tool in resp.tools
    ]
    print("可用工具：", [t["function"]["name"] for t in tools])

    # 4. 主交互循环
    while True:
        print("\n请输入你的加法问题（如：5 加 7 是多少？或'退出'）：")
        user_input = input("> ")
        if user_input.strip().lower() == '退出':
            break
        print(f"\n📝 用户问题：{user_input}")

        # 构造对话
        messages = [
            {"role": "system", "content": "你是一个加法助手，遇到加法问题请调用工具 add，最后用自然语言回答用户。"},
            {"role": "user", "content": user_input}
        ]

        # 5. LLM 与工具调用循环
        iteration = 0
        while True:
            iteration += 1
            print(f"\n🔄 第 {iteration} 次 LLM 调用...")
            # 调用 LLM
            response = client.chat.completions.create(
                model="qwen-plus",
                messages=messages,
                tools=tools,
                tool_choice="auto"
            )
            message = response.choices[0].message
            messages.append(message)

            # 检查是否有工具调用
            if not message.tool_calls:
                print(f"\n✅ LLM 最终回答：")
                print(f"\nAI 回答：\n {message.content}")
                break

            # 6. 执行工具调用
            for tool_call in message.tool_calls:
                args = json.loads(tool_call.function.arguments)
                print(f"\n🔧 调用工具：{tool_call.function.name}")
                print(f"📥 工具参数：{args}")
                result = await session.call_tool(tool_call.function.name, args)
                print(f"📤 工具返回结果：{result}")
                # 将工具结果加入对话历史
                messages.append({"role": "tool", "content": str(result), "tool_call_id": tool_call.id})

        await session.__aexit__(None, None, None)
        await transport.__aexit__(None, None, None)
        print(">>> 客户端已关闭")

if __name__ == "__main__":
    asyncio.run(main())

4. 第三步：配置环境变量

创建或修改 .env 文件：

# 通义千问 API 配置（使用阿里云 DashScope）
QWEN_API_KEY=your-api-key-here
QWEN_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1

5. 第四步：运行程序

5.1 方式一：两个终端分别运行

终端 1：启动服务器

cd hello-world
uv run python server.py

终端 2：运行客户端

cd hello-world
QWEN_API_KEY="your-api-key"
QWEN_BASE_URL="https://dashscope.aliyuncs.com/compatible-mode/v1"
uv run python client-qwen.py

5.2 方式二：非交互式测试

cd hello-world
timeout 60 bash -c 'echo -e "15 加 8 等于多少？\n退出" | uv run python client-qwen.py'

6. 执行流程示例

当用户输入 '15 加 8 等于多少？' 时：

📝 用户问题：15 加 8 等于多少？
🔄 第 1 次 LLM 调用...
🔧 调用工具：add
📥 工具参数：{'a': 15, 'b': 8}
📤 工具返回结果：meta=None content=[TextContent(type='text', text='23', annotations=None)] isError=False
🔄 第 2 次 LLM 调用...
✅ LLM 最终回答：
AI 回答：15 加 8 等于 23。

7. 关键要点

7.1 异步编程

MCP 使用 asyncio，所有网络操作都是异步的：

async def main():
    # 异步操作
    await session.initialize()
    result = await session.call_tool(...)

7.2 工具定义

简单易用的装饰器风格：

@mcp.tool()
def my_tool(param1: int, param2: str) -> str:
    """Tool description"""
    return f"Result: {param1}{param2}"

7.3 工具调用链

LLM 根据需要多次调用工具，直到得到最终答案：

用户输入 → LLM 分析 → 调用工具 → 获得结果 → LLM 再次分析 → 最终回答

7.4 消息历史

保持对话历史以便 LLM 理解上下文：

messages = [
    {"role": "system", "content": "..."},
    {"role": "user", "content": "..."},
    {"role": "assistant", "content": "..."},
    {"role": "tool", "content": "..."},
]

8. 扩展应用

8.1 添加更多工具

@mcp.tool()
def multiply(a: int, b: int) -> int:
    """Multiply two numbers"""
    return a * b

@mcp.tool()
def divide(a: float, b: float) -> float:
    """Divide two numbers"""
    if b == 0:
        raise ValueError("Cannot divide by zero")
    return a / b

8.2 添加资源

@mcp.resource("user://{user_id}")
def get_user_info(user_id: str) -> str:
    """Get user information"""
    return f"User {user_id} information"

8.2.1 什么是资源（Resource）？

@mcp.resource() 装饰器用于定义一个可以根据参数返回不同数据的接口。资源是不同于工具的数据取…

资源 vs 工具的对比：

8.2.2 URI 模式详解

@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
    """Get a personalized greeting"""
    return f"Hello, {name}!"

语法分解：

工作流程：

客户端请求 greeting://Alice
↓ MCP 框架识别 URI 模式
↓ 提取参数 name = "Alice"
↓ 调用函数 get_greeting("Alice")
↓ 返回 "Hello, Alice!" 给客户端

8.2.3 常见资源示例

示例 1：用户信息资源

@mcp.resource("user://{user_id}")
def get_user_info(user_id: str):
    users = {"1": "Alice", "2": "Bob"}
    return users.get(user_id, "Not found")
# 客户端调用：await session.read_resource("user://1")
# 返回：Alice

示例 2：文件资源（多参数）

@mcp.resource("file://{folder}/{filename}")
def read_file(folder: str, filename: str) -> str:
    # 注意：参数名必须与 URI 中的占位符一致
    path = f"{folder}/{filename}"
    try:
        with open(path, 'r') as f:
            return f.read()
    except FileNotFoundError:
        return "File not found"
# 客户端调用：await session.read_resource("file://docs/readme.txt")
# 返回：文件内容

示例 3：API 文档资源

@mcp.resource("docs://api/{version}")
def get_api_docs(version: str) -> str:
    docs = {"v1": "API v1: 支持基础功能", "v2": "API v2: 新增高级功能"}
    return docs.get(version, "Version not found")

示例 4：多参数资源

@mcp.resource("product://{category}/{product_id}")
def get_product(category: str, product_id: str) -> str:
    # 访问步骤：product://electronics/12345
    return f"Product {product_id} from {category}"

8.2.4 客户端中使用资源

# 方法 1：直接读取资源
result = await session.read_resource("greeting://Alice")
print(result)
# 输出：Hello, Alice!

# 方法 2：列出所有资源
resources = await session.list_resources()
for resource in resources.resources:
    print(f"Resource: {resource.uri} - {resource.description}")

8.2.5 资源 URI 命名最佳实践

✅ 好的设计：

@mcp.resource("user://{user_id}")
@mcp.resource("file://{path}/{filename}")
@mcp.resource("docs://api/{version}")

❌ 避免的做法：

@mcp.resource("res://{id}")  # ❌ 不清楚的协议名
@mcp.resource("user://{user_id}")
def get_user(uid: str):      # ❌ 参数名不匹配！
    pass

8.3 自定义 System Prompt

system_prompt = """
你是一个数学助手。
当用户问到加法、减法、乘法、除法时，请调用相应的工具。
最后用清晰的中文回答用户的问题。
"""
messages = [
    {"role": "system", "content": system_prompt},
    {"role": "user", "content": user_input}
]

9. 常见问题

9.1 Q1: 如何调试 MCP 服务器？

使用日志输出：

import logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

@mcp.tool()
def my_tool(x: int):
    logger.info(f"Tool called with x={x}")
    return x * 2

9.2 Q2: 如何处理工具调用错误？

try:
    result = await session.call_tool(tool_name, args)
except Exception as e:
    print(f"Tool call failed: {e}")
    # 错误恢复逻辑

9.3 Q3: 如何支持多轮对话？

保持 messages 列表，每次交互都添加新消息：

while True:
    user_input = input("> ")
    messages.append({"role": "user", "content": user_input})
    # ... 处理响应，添加到 messages ...

10. 性能优化建议

1. 连接复用 - 不要频繁创建/销毁连接
2. 超时设置 - 为 LLM 调用设置合理超时
3. 错误重试 - 实现指数退避重试机制
4. 日志级别 - 生产环境减少日志输出

11. 总结

通过 MCP，你可以轻松构建强大的 AI 应用，让 LLM 能够访问和使用自定义工具。核心步骤是：

1. 定义工具（Server）
2. 连接到服务器（Client）
3. 让 LLM 自动调用工具
4. 处理工具结果并返回给用户

希望这个教程能帮助你开始 MCP 的学习之旅！

12. 参考资源

• MCP 官方文档
• FastMCP 框架
• OpenAI API 文档
• 本项目代码