Python MCP Client实战：为AI助手构建标准化工具与资源连接
在AI应用开发领域，工具调用是实现AI与外部系统交互的核心技术。其基本原理是通过定义标准化的通信协议，让AI模型能够安全、可靠地执行外部函数或访问数据源。这项技术的核心价值在于解决了传统AI应用开发中工具集成的碎片化问题，通过统一的接口规范，显著提升了开发效率和系统可维护性。在实际应用场景中，开发者可以为AI助手（如Claude、GPTs）快速集成数据库查询、API调用、文件操作等能力，构建出功能
1. 项目概述：MCP Client 是什么，以及为什么你需要关注它
如果你最近在关注AI应用开发，特别是想让你的AI助手（比如Claude、GPTs）能够“动手”操作外部工具和数据源，那么你很可能已经听说过“Model Context Protocol”，也就是MCP。而 theailanguage/mcp_client 这个项目，就是一个用Python实现的、开箱即用的MCP客户端库。简单来说，它为你提供了一个标准化的“接线板”，让你能轻松地把各种工具（数据库、API、文件系统，甚至是你的智能家居开关）连接到像Claude Desktop这样的AI应用上，极大地扩展了AI助手的能力边界。
我最初接触MCP，是因为厌倦了为每一个AI项目重复编写类似的“工具调用”代码。无论是让AI查询数据库，还是调用一个天气API，底层逻辑都大同小异：处理请求、调用工具、格式化响应、处理错误。MCP的出现，就是为了解决这个碎片化的问题。它定义了一套AI模型与外部工具之间通信的通用协议，而 mcp_client 就是这个协议在Python侧的“官方”实现之一。它让你可以专注于定义你的工具能做什么（即实现“Server”端），而无需操心如何与AI应用建立连接、传输数据这些底层细节。对于开发者而言，这意味着更快的开发速度、更一致的接口和更好的可维护性。
这个项目适合任何想要为AI助手构建自定义能力的开发者，无论你是想集成内部业务系统，还是想打造一个拥有独特技能的个性化AI伴侣。接下来，我会带你从零开始，深入拆解如何使用 mcp_client ，并分享我在实际项目中踩过的坑和总结出的最佳实践。
2. MCP 核心概念与架构深度解析
在动手写代码之前，我们必须先吃透MCP的几个核心概念。这能帮你理解 mcp_client 在整个体系中的位置，以及你该如何设计你的工具。
2.1 MCP 的三层架构：Client, Server 与 Transport
MCP的架构非常清晰，可以理解为“客户端-服务器”模式，中间由“传输层”连接。
-
Server（工具提供方） ：这是你将要编写的主要部分。一个MCP Server定义了一系列“工具”（Tools）和“资源”（Resources）。 工具 就是AI可以调用的函数，比如
search_web,execute_sql。 资源 则是AI可以读取的静态或动态内容，比如一个配置文件、一个数据库表的结构描述。你的业务逻辑就封装在这里。 -
Client（AI应用方） ：这就是
theailanguage/mcp_client扮演的角色。它负责与Server建立连接，向AI模型（如Claude）宣告Server提供了哪些工具和资源，并在AI发出指令时，调用对应的Server工具，然后将结果返回给AI。Claude Desktop、Cursor等AI应用内部就集成了一个MCP Client。 -
Transport（传输层） ：这是Client和Server通信的通道。MCP支持几种方式：
- stdio（标准输入输出） ：最常见的方式，Server作为一个独立的进程启动，通过命令行管道与Client通信。部署简单，适合本地开发。
- SSE（Server-Sent Events） ：基于HTTP的传输方式，Server作为一个HTTP服务运行，Client通过HTTP请求与之通信。更适合远程部署和云原生环境。
mcp_client 库的核心价值在于，它帮你实现了Client侧与Server通信的所有协议细节。你不需要去手动拼接JSON-RPC格式的消息，也不需要处理复杂的生命周期管理（如初始化、列表工具、调用工具）。你只需要告诉 mcp_client 你的Server在哪里（通过什么Transport），它就会帮你打理好一切。
2.2 协议核心：工具（Tools）与资源（Resources）的设计哲学
理解Tools和Resources的区别，是设计一个好用的MCP Server的关键。
Tools 是“动词” ，代表一个可执行的动作。每个Tool必须有：
name: 唯一标识符，如get_weather。description: 给AI看的自然语言描述， 至关重要 。AI根据这个描述决定是否以及如何调用它。描述应清晰说明功能、输入参数和输出。inputSchema: 定义输入参数的JSON Schema。这告诉AI需要提供什么信息，以及信息的类型（字符串、数字、对象等）。
Resources 是“名词” ，代表可被读取的信息。它们可以是：
uri: 类似URL的唯一标识符，如file:///etc/config.yaml或dynamic://system/status。mimeType: 资源的媒体类型，如text/plain,application/json，帮助AI正确解析内容。name和description: 方便AI理解资源是什么。
Resources的一个强大之处在于支持“模板化URI”。例如，你可以定义一个资源模板 weather://{city} ，当AI需要某个城市的天气时，它可以请求 weather://beijing 这个具体的资源，你的Server再动态生成内容。
实操心得 ：在设计Tool的
description时，要站在AI的角度思考。不要只写“查询用户”，而应该写“根据用户ID查询用户的姓名、邮箱和注册时间。输入参数user_id为字符串类型”。越精确的描述，AI调用的准确率越高。对于Resources，如果你的内容很大，考虑通过text/plain提供摘要，并通过Tool提供详情查询，避免一次性加载过多数据拖慢响应。
3. 使用 mcp_client 构建你的第一个MCP Server
理论讲完了，我们动手实现一个最简单的MCP Server。这个Server将提供一个查询服务器当前时间的工具。
3.1 环境准备与项目初始化
首先，确保你的Python环境是3.8或更高版本。然后创建一个新的项目目录并安装必要的包。
mkdir my_first_mcp_server
cd my_first_mcp_server
python -m venv venv
# 在Windows上使用 `venv\Scripts\activate`
source venv/bin/activate
pip install mcp
这里我们安装的是 mcp 包，它是 theailanguage/mcp_client 项目发布的PyPI包名。这个包同时包含了开发Server所需的工具（ mcp.server ）和一些辅助功能。
3.2 实现一个简单的时间查询Server
创建一个名为 server.py 的文件。
# server.py
import asyncio
from datetime import datetime
from mcp.server import Server
from mcp.server.models import InitializationOptions
import mcp.server.stdio
from mcp.shared.models import Tool, TextContent
# 1. 创建Server实例
server = Server("simple-time-server")
# 2. 使用装饰器注册一个Tool
@server.list_tools()
async def handle_list_tools():
# 返回Server提供的所有工具列表
return [
Tool(
name="get_current_time",
description="获取服务器当前的日期和时间。无需任何参数。",
inputSchema={
"type": "object",
"properties": {}, # 无输入参数
"required": []
}
)
]
# 3. 实现Tool的执行逻辑
@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list[TextContent]:
if name == "get_current_time":
current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# 返回的内容必须包装在 TextContent 列表中
return [TextContent(type="text", text=f"当前服务器时间是：{current_time}")]
# 如果收到未知的工具名，可以抛出错误
raise ValueError(f"未知的工具：{name}")
# 4. 主异步函数，启动Server
async def main():
# 使用stdio传输层运行server
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="simple-time-server",
server_version="0.1.0"
)
)
if __name__ == "__main__":
asyncio.run(main())
这段代码做了以下几件事：
- 创建了一个名为
simple-time-server的Server实例。 - 定义了
handle_list_tools函数，并用@server.list_tools()装饰它。当Client询问“你有什么工具？”时，这个函数会被调用，返回我们定义的get_current_time工具的信息。 - 定义了
handle_call_tool函数，并用@server.call_tool()装饰它。当Client（代表AI）调用get_current_time工具时，这个函数被执行，它获取当前时间并格式化返回。 main函数建立了stdio传输层，并启动Server。
3.3 配置Claude Desktop进行测试
要让这个Server被Claude Desktop使用，你需要创建一个配置文件。
-
找到Claude Desktop的配置目录 ：
- macOS :
~/Library/Application Support/Claude/claude_desktop_config.json - Windows :
%APPDATA%\Claude\claude_desktop_config.json - Linux :
~/.config/Claude/claude_desktop_config.json
- macOS :
-
编辑（或创建）
claude_desktop_config.json文件 ：{ "mcpServers": { "simple-time-server": { "command": "python", "args": [ "/ABSOLUTE/PATH/TO/YOUR/server.py" // 替换为你的 server.py 的绝对路径 ], "env": { "PYTHONUNBUFFERED": "1" } } } }关键点 ：
args中的路径必须是 绝对路径 。PYTHONUNBUFFERED环境变量确保Python的输出能立即被Claude Desktop接收到。 -
重启Claude Desktop 。
重启后，在Claude的聊天界面，你应该能看到一个微小的插件图标（通常是个拼图块或螺丝刀图标）。点击它，如果配置成功，你会看到 simple-time-server 已加载。现在，你可以对Claude说：“请调用 get_current_time 工具。” 或者更自然地说：“现在服务器时间是多少？” Claude应该能识别并调用你的工具，返回时间信息。
注意事项 ：第一次配置时最常见的失败原因是路径错误或Python环境问题。确保
command指定的python命令在你系统的PATH中，并且它指向的是安装了mcp包的那个虚拟环境下的Python。一个调试方法是，先在终端里用同样的命令手动运行你的server.py，看脚本是否能正常启动而不报错。
4. 构建功能完整的MCP Server：以数据库查询为例
一个只会报时的Server显然不够看。我们来构建一个更实用的Server：一个可以查询SQLite数据库的MCP Server。这将涉及更复杂的Tool定义、错误处理以及Resources的使用。
4.1 设计数据库工具集
假设我们有一个 users 表，结构如下：
CREATE TABLE users (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
email TEXT UNIQUE NOT NULL,
age INTEGER
);
我们想提供三个工具：
list_users: 列出所有用户（可分页）。get_user_by_id: 根据ID查询特定用户。search_users_by_name: 根据姓名模糊搜索用户。
同时，我们提供一个资源 database://schema ，让AI可以读取数据库的表结构。
4.2 实现数据库Server
创建 db_server.py 。我们需要额外安装 aiosqlite 来进行异步数据库操作。
pip install aiosqlite
# db_server.py
import asyncio
import aiosqlite
from typing import Any, Optional
from mcp.server import Server
from mcp.server.models import InitializationOptions
import mcp.server.stdio
from mcp.shared.models import Tool, TextContent, Resource
server = Server("user-database-server")
DB_PATH = "example.db" # 你的数据库文件路径
# --- 初始化时创建示例数据库（仅演示用）---
async def init_database():
async with aiosqlite.connect(DB_PATH) as db:
await db.execute("""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
email TEXT UNIQUE NOT NULL,
age INTEGER
)
""")
# 插入一些示例数据
await db.executemany(
"INSERT OR IGNORE INTO users (id, name, email, age) VALUES (?, ?, ?, ?)",
[(1, 'Alice', 'alice@example.com', 30),
(2, 'Bob', 'bob@example.com', 25),
(3, 'Charlie', 'charlie@example.com', 35)]
)
await db.commit()
# --- 注册工具列表 ---
@server.list_tools()
async def handle_list_tools():
return [
Tool(
name="list_users",
description="列出所有用户，支持分页。参数：page（页码，从1开始，默认1），per_page（每页条数，默认10）。",
inputSchema={
"type": "object",
"properties": {
"page": {"type": "integer", "description": "页码，从1开始", "minimum": 1, "default": 1},
"per_page": {"type": "integer", "description": "每页数量", "minimum": 1, "maximum": 100, "default": 10}
},
"required": []
}
),
Tool(
name="get_user_by_id",
description="根据用户ID获取用户的详细信息。",
inputSchema={
"type": "object",
"properties": {
"user_id": {"type": "integer", "description": "用户ID"}
},
"required": ["user_id"]
}
),
Tool(
name="search_users_by_name",
description="根据姓名进行模糊搜索。",
inputSchema={
"type": "object",
"properties": {
"name_query": {"type": "string", "description": "姓名搜索关键词"}
},
"required": ["name_query"]
}
)
]
# --- 注册资源列表 ---
@server.list_resources()
async def handle_list_resources():
# 提供一个描述数据库模式的资源
return [
Resource(
uri="database://schema",
name="database-schema",
description="用户数据库的表结构定义。",
mimeType="text/plain"
)
]
# --- 处理读取资源的请求 ---
@server.read_resource()
async def handle_read_resource(uri: str) -> list[TextContent]:
if uri == "database://schema":
schema_text = """
Table: users
- id: INTEGER (PRIMARY KEY)
- name: TEXT (NOT NULL)
- email: TEXT (NOT NULL, UNIQUE)
- age: INTEGER
"""
return [TextContent(type="text", text=schema_text)]
raise ValueError(f"未知的资源URI: {uri}")
# --- 处理工具调用 ---
@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list[TextContent]:
async with aiosqlite.connect(DB_PATH) as db:
db.row_factory = aiosqlite.Row # 以字典形式返回行
if name == "list_users":
page = arguments.get("page", 1)
per_page = arguments.get("per_page", 10)
offset = (page - 1) * per_page
async with db.execute(
"SELECT * FROM users LIMIT ? OFFSET ?",
(per_page, offset)
) as cursor:
rows = await cursor.fetchall()
users = [dict(row) for row in rows]
# 获取总数用于分页信息
async with db.execute("SELECT COUNT(*) as total FROM users") as cursor:
total_row = await cursor.fetchone()
total = total_row["total"]
result_text = f"第{page}页，共{per_page}条/页，总计{total}条记录：\n"
for user in users:
result_text += f"- ID:{user['id']}, 姓名:{user['name']}, 邮箱:{user['email']}, 年龄:{user['age']}\n"
return [TextContent(type="text", text=result_text)]
elif name == "get_user_by_id":
user_id = arguments["user_id"]
async with db.execute(
"SELECT * FROM users WHERE id = ?",
(user_id,)
) as cursor:
row = await cursor.fetchone()
if row:
user = dict(row)
return [TextContent(type="text", text=f"找到用户：{user}")]
else:
return [TextContent(type="text", text=f"未找到ID为 {user_id} 的用户。")]
elif name == "search_users_by_name":
query = f"%{arguments['name_query']}%"
async with db.execute(
"SELECT * FROM users WHERE name LIKE ?",
(query,)
) as cursor:
rows = await cursor.fetchall()
users = [dict(row) for row in rows]
if users:
result_text = f"找到 {len(users)} 个匹配的用户：\n"
for user in users:
result_text += f"- {user['name']} (ID: {user['id']}, 邮箱: {user['email']})\n"
else:
result_text = f"未找到姓名包含 '{arguments['name_query']}' 的用户。"
return [TextContent(type="text", text=result_text)]
else:
raise ValueError(f"未知的工具：{name}")
# --- 主函数 ---
async def main():
await init_database() # 初始化数据库
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="user-database-server",
server_version="1.0.0"
)
)
if __name__ == "__main__":
asyncio.run(main())
这个Server的复杂度显著提升，它展示了：
- 带参数的Tool ：
list_users工具包含了有默认值、有范围限制的参数。 - Resource的提供 ：通过
database://schema资源，AI可以在需要了解数据库结构时主动读取，这比在Tool描述里写死更灵活。 - 完整的数据库操作与错误处理 ：使用
aiosqlite进行异步操作，并处理了查询无结果的情况。
4.3 进阶：动态资源与模板化URI
让我们更进一步，为每个用户创建一个动态资源，比如 user://{id} 。这样AI可以直接说“读取 user://1 的信息”，而不必调用工具。
修改 handle_list_resources 和 handle_read_resource ：
# ... 前面的代码不变 ...
@server.list_resources()
async def handle_list_resources():
resources = [
Resource(
uri="database://schema",
name="database-schema",
description="用户数据库的表结构定义。",
mimeType="text/plain"
)
]
# 动态添加用户资源（这里只是示例，实际可能根据数据库内容动态生成列表）
# 更常见的做法是在 read_resource 时根据URI动态判断
# 这里我们演示一个模板化的资源
resources.append(
Resource(
uri="user://{id}",
name="user-details",
description="根据用户ID获取用户详细信息。",
mimeType="application/json"
)
)
return resources
@server.read_resource()
async def handle_read_resource(uri: str) -> list[TextContent]:
if uri == "database://schema":
# ... 返回模式 ...
pass
elif uri.startswith("user://"):
# 解析用户ID
try:
user_id = int(uri.split("://")[1])
except ValueError:
raise ValueError(f"无效的用户资源URI: {uri}")
async with aiosqlite.connect(DB_PATH) as db:
db.row_factory = aiosqlite.Row
async with db.execute(
"SELECT * FROM users WHERE id = ?",
(user_id,)
) as cursor:
row = await cursor.fetchone()
if row:
user = dict(row)
# 以JSON格式返回
import json
return [TextContent(type="text", text=json.dumps(user, indent=2, ensure_ascii=False))]
else:
return [TextContent(type="text", text=f"{{ \"error\": \"User {user_id} not found\" }}")]
else:
raise ValueError(f"未知的资源URI: {uri}")
现在，AI就可以通过 user://1 这样的URI直接请求资源了。这体现了MCP的灵活性：既可以通过Tools执行动作，也可以通过Resources获取信息。
5. 生产环境部署与性能优化考量
当你的MCP Server从玩具变成真正要服务生产环境时，有几个关键点需要考虑。
5.1 传输层选择：Stdio vs. SSE
- Stdio ：简单，适合本地开发、桌面集成或与单个AI应用进程一对一绑定。但它不适合高并发或需要被多个Client同时访问的场景。
- SSE (Server-Sent Events) ：基于HTTP，意味着你的Server可以作为一个独立的Web服务部署。这带来了巨大优势：
- 可扩展性 ：可以部署在云上，被多个Claude Desktop实例甚至其他兼容MCP的客户端访问。
- 独立性 ：Server的生命周期不再依赖Client进程。你可以独立重启、升级Server。
- 可观测性 ：可以利用成熟的HTTP监控、日志、负载均衡工具。
使用SSE需要修改Server的启动方式。 mcp 库提供了 mcp.server.sse 模块来简化这一过程。你需要一个像FastAPI或aiohttp这样的ASGI/HTTP框架。
# sse_server.py (示例框架)
from fastapi import FastAPI
from mcp.server.sse import SseServerTransport
from mcp.server import Server
import asyncio
# ... 你的工具和资源注册代码与之前类似 ...
app = FastAPI()
server = Server("my-sse-server")
# 创建SSE传输层
transport = SseServerTransport("/messages")
# 将传输层连接到server
@app.on_event("startup")
async def startup_event():
asyncio.create_task(server.connect(transport))
# 暴露SSE端点
@app.get("/sse")
async def handle_sse(request: Request):
return await transport.handle_request(request)
# ... 注册你的list_tools, call_tool等处理函数 ...
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
然后在Claude Desktop配置中，将 command 方式改为 url 方式：
{
"mcpServers": {
"my-remote-server": {
"url": "http://localhost:8000/sse"
}
}
}
5.2 错误处理与日志记录
生产环境的Server必须有健壮的错误处理。
- Tool调用错误 ：在
handle_call_tool内部用try...except包裹核心逻辑，返回清晰的错误信息给AI，而不是让整个Server崩溃。@server.call_tool() async def handle_call_tool(name: str, arguments: dict) -> list[TextContent]: try: # ... 你的业务逻辑 ... if some_error_condition: # 返回结构化的错误信息给AI return [TextContent(type="text", text=f"错误：操作失败，原因：{reason}")] except aiosqlite.Error as e: # 记录到日志 logger.error(f"数据库操作失败: {e}") # 返回用户友好的错误信息 return [TextContent(type="text", text="数据库服务暂时不可用，请稍后再试。")] except Exception as e: logger.exception(f"处理工具 {name} 时发生未知错误") return [TextContent(type="text", text="服务器内部错误。")] - 日志记录 ：使用标准的
logging模块，将日志输出到文件或日志收集系统（如ELK、Loki）。记录Server的启动、关闭、每个工具的调用请求和响应（注意脱敏敏感数据）、错误信息等。这对于调试和监控至关重要。
5.3 安全性与权限控制
这是企业级应用无法回避的话题。
- 输入验证与消毒 ：永远不要相信来自AI的输入。即使有
inputSchema，在Tool的实现代码中，也要对arguments进行二次验证。特别是涉及数据库查询时，要使用参数化查询（如上例所示），绝对 禁止 字符串拼接，以防止SQL注入。 - 权限模型 ：你的Server可能需要服务不同权限的用户。MCP协议本身不包含身份验证。权限控制需要在更高层实现：
- 方案A（SSE模式） ：在HTTP层实现认证（如API Key、JWT Token）。Claude Desktop配置中可以添加自定义Headers（如果其Client支持）。Server端验证Token并关联用户权限，在Tool执行前检查。
- 方案B（Stdio模式） ：权限信息较难传递。一种思路是运行多个不同权限配置的Server实例，在Claude Desktop中按需切换。另一种是在Tool内部实现一个简单的基于会话或环境的权限检查（但这不够安全）。
- 敏感信息 ：不要在Tool的
description或Resource内容中泄露敏感信息（如数据库连接字符串、内部API密钥）。使用环境变量或安全的配置管理工具来存储这些信息。
实操心得 ：对于内部工具，开始时可以基于“信任边界”来简化权限。例如，假设能访问这台电脑并运行Claude Desktop的人就是授权用户。但随着工具能力的扩大（如能删除数据、调用付费API），必须引入明确的权限控制。一个简单的起步方法是，为每个Tool增加一个
required_permission参数，并在Server启动时从环境变量加载当前用户的权限列表，在调用时进行匹配。
6. 调试、监控与常见问题排查
开发MCP Server时，你肯定会遇到各种问题。这里总结一套调试流程和常见坑位。
6.1 调试流程与工具
- 首先，独立测试你的Server逻辑 ：在集成到MCP之前，先写一个简单的Python脚本，直接调用你的
handle_call_tool函数，确保核心业务逻辑正确。这能排除掉MCP协议本身带来的复杂度。 - 使用MCP Inspector进行协议级调试 ：Anthropic官方提供了一个非常有用的工具叫 MCP Inspector 。它是一个独立的MCP Client，可以连接到你的Server，并图形化地展示所有可用的Tools和Resources，还能手动调用它们并查看原始请求/响应消息。这是调试协议通信问题的利器。
# 通常可以通过npx运行 npx @modelcontextprotocol/inspector <你的server启动命令> # 例如对于我们的stdio server npx @modelcontextprotocol/inspector python /path/to/your/server.py - 查看Claude Desktop日志 ：当Claude Desktop无法加载你的Server时，查看其日志文件能找到线索。日志位置通常与配置文件在同一目录下。
- 在Server中添加详细日志 ：在
list_tools,call_tool,read_resource等函数的入口和出口添加日志记录，打印接收到的参数和返回的结果。
6.2 常见问题速查表
| 问题现象 | 可能原因 | 排查步骤 |
|---|---|---|
| Claude Desktop中看不到Server图标 | 配置文件错误、Server启动失败 | 1. 检查配置文件JSON格式是否正确。 2. 检查 command 和 args 路径是否正确、可执行。 3. 在终端手动运行配置中的命令，看Server是否能正常启动并持续运行（不立刻退出）。 4. 检查Claude Desktop日志。 |
| Server图标显示为“错误”或“断开” | Server进程启动后崩溃、协议通信失败 | 1. 在Server代码开头添加 print("Server starting...") ，确认进程已启动。 2. 使用MCP Inspector连接，查看具体的错误信息。 3. 检查Server代码是否有未捕获的异常，特别是在初始化阶段。 |
| AI无法调用某个工具 | Tool的 description 不清晰、 inputSchema 不匹配 |
1. 在MCP Inspector中查看该Tool的定义是否完整。 2. 检查 inputSchema 的 required 字段和 properties 定义是否与AI调用时传递的参数匹配。 3. 优化Tool的 description ，使其对AI更友好。 |
| 工具调用超时或无响应 | Server端工具执行时间过长、死锁 | 1. 确保你的工具函数是异步的（ async ），并且内部没有执行同步的耗时操作（如 time.sleep ）。应使用 asyncio.sleep 。 2. 为数据库查询、网络请求设置合理的超时时间。 3. 在工具函数中添加日志，记录开始和结束时间。 |
| 返回的内容AI无法理解 | 返回格式不符合预期、内容过于冗长或非结构化 | 1. 确保返回的是 List[TextContent] 。 2. 返回给AI的文本应尽量简洁、结构化。对于复杂数据，考虑使用Markdown表格或JSON格式，AI能更好地解析。 3. 避免返回巨大的文本块，考虑分页或通过Resource提供详情。 |
6.3 性能优化要点
- 连接池 ：对于数据库、HTTP客户端等，使用连接池而不是每次调用都创建新连接。
- 异步无处不在 ：确保所有I/O操作（数据库、网络、文件）都使用异步库（如
aiosqlite,aiohttp,asyncpg），避免阻塞事件循环。 - 缓存 ：对于不常变化的数据（如数据库模式、配置信息），可以在Server内存中缓存，避免每次
read_resource都重复查询。 - 精简响应 ：AI处理长文本会消耗更多token和上下文。只返回必要的信息。例如，查询100条记录时，先返回摘要和首尾几条，并提供通过Tool获取更多详情的选项。
构建一个稳定、高效、安全的MCP Server是一个迭代的过程。从最简单的“Hello World”开始，逐步添加功能、完善错误处理、引入安全措施，最终将其部署到生产环境，你会深刻体会到MCP协议和 mcp_client 这类库在连接AI与现实世界中的强大威力。它不仅仅是让AI多了一个功能，更是为AI应用构建了一个标准化、可扩展的“手”和“眼”。
欢迎加入DeepSeek 技术社区。在这里，你可以找到志同道合的朋友，共同探索AI技术的奥秘。
更多推荐
7
0- 0
已为社区贡献50条内容
所有评论(0)