PythonAI算法
Python MCP 工具开发入门：Server、Client 与 LLM 集成
介绍如何使用 Python 构建 MCP（Model Context Protocol）服务器与客户端，实现与大语言模型（LLM）的集成。内容包括 MCP 核心概念、环境配置、代码示例（Server/Client）、工具与资源定义、以及多轮对话流程。通过 FastMCP 框架简化开发，支持自定义工具调用及资源读取，帮助开发者快速搭建 AI 应用。
介绍如何使用 Python 构建 MCP（Model Context Protocol）服务器与客户端，实现与大语言模型（LLM）的集成。内容包括 MCP 核心概念、环境配置、代码示例（Server/Client）、工具与资源定义、以及多轮对话流程。通过 FastMCP 框架简化开发，支持自定义工具调用及资源读取，帮助开发者快速搭建 AI 应用。
Model Context Protocol (MCP) 是一个标准化协议，允许应用程序与大语言模型（LLM）进行安全、结构化的交互。通过 MCP，你可以：
定义工具、资源和提示词，通过 stdio 或其他传输方式提供给客户端。
连接到 MCP 服务器，获取工具列表，调用工具，并与 LLM 集成。
服务器暴露给 LLM 的可调用函数，LLM 可以根据用户需求调用这些工具。
hello-world/
├── server.py # MCP 服务器定义
├── client-qwen.py # 使用 Qwen LLM 的客户端
├── client.py # 基础客户端
├── pyproject.toml # 项目配置
└── uv.lock # 依赖锁定文件
cd hello-world
uv sync
主要依赖：
mcp[cli]>=1.6.0 - MCP 框架openai>=1.75.0 - OpenAI 兼容的 LLM 客户端python-dotenv>=1.1.0 - 环境变量管理创建 server.py：
from mcp.server.fastmcp import FastMCP
# 创建一个 MCP 服务器实例
mcp = FastMCP("Demo")
# 定义一个工具：两数相加
@mcp.tool()
def add(a: int, b: int) -> int:
"""Add two numbers"""
return a + b
# 定义一个资源：个性化问候
@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
"""Get a personalized greeting"""
return f"Hello, {name}!"
# 启动服务器
if __name__ == "__main__":
mcp.run("stdio")
核心概念解析：
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())
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 tools])
:
()
user_input = ()
user_input.strip().lower() == :
()
messages = [
{: , : },
{: , : user_input}
]
iteration =
:
iteration +=
()
response = client.chat.completions.create(
model=,
messages=messages,
tools=tools,
tool_choice=
)
message = response.choices[].message
messages.append(message)
message.tool_calls:
()
()
tool_call message.tool_calls:
args = json.loads(tool_call.function.arguments)
()
()
result = session.call_tool(tool_call.function.name, args)
()
messages.append({: , : (result), : tool_call.})
session.__aexit__(, , )
transport.__aexit__(, , )
()
__name__ == :
asyncio.run(main())
创建或修改 .env 文件：
# 通义千问 API 配置（使用阿里云 DashScope）
QWEN_API_KEY=your-api-key-here
QWEN_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1
终端 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
cd hello-world
timeout 60 bash -c 'echo -e "15 加 8 等于多少？\n退出" | uv run python client-qwen.py'
当用户输入 '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。
MCP 使用 asyncio，所有网络操作都是异步的：
async def main():
# 异步操作
await session.initialize()
result = await session.call_tool(...)
简单易用的装饰器风格：
@mcp.tool()
def my_tool(param1: int, param2: str) -> str:
"""Tool description"""
return f"Result: {param1}{param2}"
LLM 根据需要多次调用工具，直到得到最终答案：
用户输入 → LLM 分析 → 调用工具 → 获得结果 → LLM 再次分析 → 最终回答
保持对话历史以便 LLM 理解上下文：
messages = [
{"role": "system", "content": "..."},
{"role": "user", "content": "..."},
{"role": "assistant", "content": "..."},
{"role": "tool", "content": "..."},
]
@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
@mcp.resource("user://{user_id}")
def get_user_info(user_id: str) -> str:
"""Get user information"""
return f"User {user_id} information"
@mcp.resource() 装饰器用于定义一个可以根据参数返回不同数据的接口。资源是不同于工具的数据取…
资源 vs 工具的对比：
| 特性 | Resource（资源） | Tool（工具） |
|---|---|---|
| 用途 | 提供只读或结构化的数据 | 执行操作或计算 |
| 调用方式 | read_resource("uri://path") | call_tool("name", args) |
| 参数传递 | URI 路径参数 | 函数参数 |
| 使用场景 | 获取文件、查询数据库 | 计算、修改数据 |
@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
"""Get a personalized greeting"""
return f"Hello, {name}!"
语法分解：
| 部分 | 含义 | 说明 |
|---|---|---|
@mcp.resource() | 资源装饰器 | 定义资源的标记 |
"greeting://" | 资源协议（Scheme） | user://、file://、api:// |
{name} | 动态参数占位符 | 类似路由参数，接收不同值 |
get_greeting(name: str) | 处理函数 | 参数名必须与 URI 占位符一致 |
工作流程：
客户端请求 greeting://Alice
↓ MCP 框架识别 URI 模式
↓ 提取参数 name = "Alice"
↓ 调用函数 get_greeting("Alice")
↓ 返回 "Hello, Alice!" 给客户端
示例 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}"
# 方法 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}")
✅ 好的设计：
@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
system_prompt = """
你是一个数学助手。
当用户问到加法、减法、乘法、除法时，请调用相应的工具。
最后用清晰的中文回答用户的问题。
"""
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_input}
]
使用日志输出：
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
try:
result = await session.call_tool(tool_name, args)
except Exception as e:
print(f"Tool call failed: {e}")
# 错误恢复逻辑
保持 messages 列表，每次交互都添加新消息：
while True:
user_input = input("> ")
messages.append({"role": "user", "content": user_input})
# ... 处理响应，添加到 messages ...
通过 MCP，你可以轻松构建强大的 AI 应用，让 LLM 能够访问和使用自定义工具。核心步骤是：
希望这个教程能帮助你开始 MCP 的学习之旅！
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online