MCP构建Tools
Tools 是 MCP 中最常用的原语，让模型能够主动调用函数完成任务。这一篇深入讲怎么写好用的 Tool。
1. 基础用法
最简单的 Tool 定义：
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("tools-demo")
@mcp.tool()
def add(a: int, b: int) -> int:
"""计算两个整数的和"""
return a + b
FastMCP 会自动：
- 用函数名作为 Tool 名称
- 用 docstring 作为 Tool 描述
- 用类型注解生成 input schema
- 用返回值类型推断输出
2. Tool 描述很重要
模型靠 description 判断什么时候调用这个 Tool。描述写得好坏直接决定模型是否能正确使用。
❌ 不好的写法：
✅ 好的写法：
@mcp.tool()
def search_documents(query: str) -> str:
"""在公司知识库中搜索文档。
适用场景：用户询问公司内部规章、产品手册、技术文档时使用。
输入：自然语言搜索词，如 "请假流程"、"API 鉴权方式"。
输出：相关文档的标题和摘要列表。
"""
...
好的描述应该包含：
- Tool 的核心功能（做什么）
- 适用场景（什么时候用）
- 输入输出说明（用什么参数、返回什么）
3. 参数类型与 Schema
FastMCP 支持丰富的 Python 类型，自动转换为 JSON Schema：
from typing import Literal
from pydantic import Field
@mcp.tool()
def search(
query: str,
limit: int = 10,
sort_by: Literal["date", "relevance"] = "relevance",
tags: list[str] | None = None,
) -> dict:
"""高级搜索接口"""
return {"results": [...]}
用 Pydantic Field 增强
复杂参数可以用 pydantic.Field 加描述和约束：
from pydantic import Field
from typing import Annotated
@mcp.tool()
def book_flight(
from_city: Annotated[str, Field(description="出发城市，如 '北京'")],
to_city: Annotated[str, Field(description="到达城市")],
date: Annotated[str, Field(description="出发日期，格式 YYYY-MM-DD", pattern=r"^\d{4}-\d{2}-\d{2}$")],
passengers: Annotated[int, Field(description="乘客数", ge=1, le=9)] = 1,
) -> str:
"""预订机票"""
return f"已预订 {from_city} → {to_city}，{date}，{passengers} 人"
每个参数都会生成完整 schema，模型能精确理解每个字段。
4. 返回结构化数据
简单情况下可以返回 str、int、dict、list，FastMCP 自动处理。
更复杂的场景，用 Pydantic 模型定义返回结构：
from pydantic import BaseModel
class WeatherInfo(BaseModel):
city: str
temperature: float
condition: str
humidity: int
@mcp.tool()
def get_weather(city: str) -> WeatherInfo:
"""查询城市天气"""
return WeatherInfo(
city=city,
temperature=22.5,
condition="晴",
humidity=65,
)
5. 返回多种内容类型
MCP 支持工具返回文本、图片、嵌入式资源等多种类型：
from mcp.server.fastmcp import Image
from mcp.types import TextContent, ImageContent
@mcp.tool()
def generate_chart(data: list[float]) -> Image:
"""生成图表"""
import matplotlib.pyplot as plt
import io
plt.plot(data)
buf = io.BytesIO()
plt.savefig(buf, format="png")
plt.close()
return Image(data=buf.getvalue(), format="png")
6. 异步 Tool
如果工具需要 I/O（HTTP 请求、数据库查询），用 async def：
import httpx
@mcp.tool()
async def fetch_url(url: str) -> str:
"""获取 URL 的内容"""
async with httpx.AsyncClient() as client:
response = await client.get(url)
return response.text[:5000] # 限制返回长度
FastMCP 会自动识别异步函数并正确调度。
7. Context：访问会话上下文
Tool 可以接收一个 Context 参数，访问当前会话的能力：
from mcp.server.fastmcp import Context
@mcp.tool()
async def long_task(items: list[str], ctx: Context) -> str:
"""处理大量数据，汇报进度"""
total = len(items)
for i, item in enumerate(items):
# 上报进度（客户端可以显示进度条）
await ctx.report_progress(i + 1, total)
# 输出日志（客户端可以显示）
await ctx.info(f"正在处理: {item}")
# ... 实际处理 ...
return f"处理完成，共 {total} 项"
Context 提供了：
ctx.info()、ctx.warning()、ctx.error()：发送日志ctx.report_progress(current, total)：上报进度ctx.read_resource(uri)：在 Tool 中读取资源ctx.session：底层会话对象，高级用法
8. 错误处理
Tool 抛出异常会被 MCP 捕获并以错误形式返回给客户端：
@mcp.tool()
def divide(a: float, b: float) -> float:
"""两数相除"""
if b == 0:
raise ValueError("除数不能为 0")
return a / b
模型看到错误后通常会重试或换方法，所以错误信息要清晰：
@mcp.tool()
def get_user(user_id: int) -> dict:
"""根据 ID 查询用户"""
user = db.find_user(user_id)
if not user:
raise ValueError(f"用户 {user_id} 不存在，请确认 ID 是否正确")
return user
如果错误很常见（如未授权），可以选择返回结构化错误而不是抛异常：
@mcp.tool()
def access_file(path: str) -> dict:
"""读取文件"""
if not is_allowed(path):
return {
"success": False,
"error": "permission_denied",
"message": f"无权访问 {path}",
}
# ... 正常逻辑 ...
模型能直接看到结构化错误并在回复中处理。
9. 调用真实 API 的完整示例
下面是一个调用真实 HTTP API 的工具示例：
import httpx
from mcp.server.fastmcp import FastMCP, Context
mcp = FastMCP("github-mcp-demo")
GITHUB_API = "https://api.github.com"
@mcp.tool()
async def get_repo_info(owner: str, repo: str, ctx: Context) -> dict:
"""获取 GitHub 仓库的基本信息。
适用场景：用户询问某个 GitHub 仓库的星数、描述、主要语言等元数据。
"""
await ctx.info(f"正在获取 {owner}/{repo}")
async with httpx.AsyncClient() as client:
response = await client.get(f"{GITHUB_API}/repos/{owner}/{repo}")
if response.status_code == 404:
raise ValueError(f"仓库 {owner}/{repo} 不存在")
if response.status_code != 200:
raise RuntimeError(f"GitHub API 请求失败: {response.status_code}")
data = response.json()
return {
"name": data["full_name"],
"description": data.get("description", ""),
"stars": data["stargazers_count"],
"forks": data["forks_count"],
"language": data.get("language", "未知"),
"url": data["html_url"],
}
@mcp.tool()
async def search_repos(keyword: str, limit: int = 5) -> list[dict]:
"""按关键词搜索 GitHub 仓库，返回前 N 个最受欢迎的结果"""
async with httpx.AsyncClient() as client:
response = await client.get(
f"{GITHUB_API}/search/repositories",
params={"q": keyword, "sort": "stars", "per_page": limit},
)
items = response.json().get("items", [])
return [{
"full_name": item["full_name"],
"description": item.get("description"),
"stars": item["stargazers_count"],
} for item in items]
if __name__ == "__main__":
mcp.run()
10. Tool 设计的最佳实践
1. 单一职责
每个 Tool 只做一件事。get_user 和 update_user 分开，不要 manage_user(action="get/update")。
2. 描述清晰
描述要让模型一眼看懂什么时候用这个 Tool。可以加适用场景示例。
3. 命名规范
用动词开头：get_*、search_*、create_*、delete_*。模型对常见命名模式更敏感。
4. 限制返回大小
避免返回超长内容把上下文撑爆。可以分页或截断：
@mcp.tool()
def list_files(path: str, max_count: int = 50) -> list[str]:
"""列出目录下的文件，最多返回 max_count 个"""
files = os.listdir(path)
return files[:max_count]
5. 提供有用的错误信息
错误消息要让模型知道接下来怎么做：
6. 警惕副作用
写入、删除、发送类操作要慎重。考虑加入 confirm 参数让模型显式确认：
@mcp.tool()
def delete_file(path: str, confirm: bool = False) -> str:
"""删除文件。confirm 必须为 True 才会真正删除。"""
if not confirm:
return "请将 confirm 设为 True 来确认删除"
os.remove(path)
return f"已删除 {path}"
总结
@mcp.tool()装饰器 + 类型注解 + docstring 即可定义工具- Tool 描述决定模型能否正确使用，要包含功能、场景、输入输出
- 用 Pydantic
Field增强参数描述和约束 - 异步 Tool 用
async def，I/O 任务务必异步 Context参数提供日志、进度上报、资源读取能力- 异常会传给客户端，错误信息要让模型知道下一步怎么办
- 设计原则：单一职责、命名规范、限制大小、警惕副作用