2.2 工具(Tools)高级用法
工具(Tools)是MCP的另一个核心概念，它们允许大语言模型执行操作和调用外部功能。在本章中，我们将深入研究工具的高级用法和最佳实践。
工具设计最佳实践
设计好的MCP工具对于提供高质量的服务至关重要。以下是一些工具设计的最佳实践：
1. 清晰的命名和描述
工具名称应该明确表达其功能，而描述应提供足够的信息以便模型理解何时使用该工具：
@mcp.tool()
def search_database(query: str, limit: int = 10) -> str:
"""搜索数据库中的记录
根据提供的查询字符串在数据库中搜索匹配的记录。
参数:
query: 搜索查询字符串
limit: 返回的最大结果数量，默认为10
返回:
匹配记录的格式化列表
"""
# 实现...
2. 职责单一
每个工具应该专注于一个特定功能，而不是尝试做多件事：
✅ 好的设计：
@mcp.tool()
def get_weather(city: str) -> str:
"""获取指定城市的天气信息"""
# 实现...
@mcp.tool()
def get_news(category: str) -> str:
"""获取指定类别的新闻"""
# 实现...
❌ 不好的设计：
@mcp.tool()
def get_weather_and_news(city: str, news_category: str) -> str:
"""获取天气和新闻信息"""
# 实现...
3. 明确的参数类型
使用类型注解明确指定参数类型，这有助于模型正确调用工具：
@mcp.tool()
def calculate_mortgage(
principal: float,
interest_rate: float,
years: int,
payment_frequency: str = "monthly"
) -> float:
"""计算抵押贷款的定期还款金额"""
# 实现...
4. 提供合理的默认值
对于可选参数，提供合理的默认值可以简化工具的使用：
@mcp.tool()
def search_products(
query: str,
category: str = "all",
min_price: float = 0.0,
max_price: float = 1000000.0,
sort_by: str = "relevance",
page: int = 1,
items_per_page: int = 20
) -> str:
"""搜索产品目录"""
# 实现...
5. 返回结构化数据
当合适时，返回结构化数据而不是纯文本字符串，这可以使模型更容易处理结果：
@mcp.tool()
def get_stock_info(symbol: str) -> dict:
"""获取股票信息"""
# 实现...
return {
"symbol": symbol,
"price": 123.45,
"change": 1.23,
"volume": 1000000,
"market_cap": 1000000000
}
参数类型和验证
MCP工具支持多种参数类型和验证方式，这有助于确保工具接收到有效的输入。
1. 基本类型
MCP支持的基本参数类型包括：
str: 字符串int: 整数float: 浮点数bool: 布尔值list: 列表dict: 字典/对象
2. 自定义类型
对于更复杂的参数，可以使用自定义类型：
from enum import Enum
from typing import List, Dict, Optional, Union
from dataclasses import dataclass
class SortOrder(Enum):
ASCENDING = "asc"
DESCENDING = "desc"
@dataclass
class SearchParams:
query: str
filters: Dict[str, str]
page: int
order: SortOrder
@mcp.tool()
def advanced_search(params: SearchParams) -> list:
"""高级搜索功能"""
# 实现...
3. 参数验证
可以使用验证装饰器或内联验证来确保参数值有效：
from mcp.validation import validate_params, Range, OneOf
@mcp.tool()
@validate_params({
"age": Range(min=0, max=120),
"country": OneOf(["美国", "中国", "英国", "德国", "法国"]),
"email": {"pattern": r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"}
})
def register_user(name: str, age: int, country: str, email: str) -> str:
"""注册新用户"""
# 实现...
或者使用内联验证：
@mcp.tool()
def transfer_money(
from_account: str,
to_account: str,
amount: float
) -> str:
"""转账操作"""
# 验证
if amount <= 0:
raise ValueError("转账金额必须大于零")
if from_account == to_account:
raise ValueError("源账户和目标账户不能相同")
# 实现...
异步工具函数
MCP支持异步工具函数，这对于需要进行I/O操作的工具特别有用。
1. 基本异步工具
@mcp.tool()
async def fetch_remote_data(url: str) -> str:
"""从远程URL获取数据"""
import aiohttp
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
text = await response.text()
return text
2. 组合多个异步操作
@mcp.tool()
async def aggregate_data(apis: List[str]) -> dict:
"""从多个API聚合数据"""
import aiohttp
import asyncio
async def fetch_single(url):
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return await response.json()
# 并行执行所有请求
tasks = [fetch_single(api) for api in apis]
results = await asyncio.gather(*tasks)
# 处理结果
combined_data = {}
for i, result in enumerate(results):
combined_data[f"api_{i}"] = result
return combined_data
3. 处理超时和取消
@mcp.tool()
async def fetch_with_timeout(url: str, timeout: float = 5.0) -> str:
"""带超时的数据获取"""
import aiohttp
import asyncio
try:
async with aiohttp.ClientSession() as session:
async with session.get(url, timeout=timeout) as response:
return await response.text()
except asyncio.TimeoutError:
return f"获取数据超时（超过{timeout}秒）"
except Exception as e:
return f"获取数据出错: {str(e)}"
返回复杂数据结构
MCP工具可以返回多种类型的复杂数据结构，以满足不同需求。
1. 结构化JSON数据
@mcp.tool()
def get_user_profile(user_id: str) -> dict:
"""获取用户资料"""
# 从数据库获取用户数据
# ...
return {
"id": user_id,
"name": "张三",
"email": "zhangsan@example.com",
"age": 30,
"address": {
"city": "北京",
"street": "朝阳区xxx路"
},
"interests": ["编程", "读书", "旅行"]
}
2. 表格数据
from mcp.types import TableContent
@mcp.tool()
def query_sales_data(start_date: str, end_date: str) -> TableContent:
"""查询销售数据，返回表格格式"""
# 获取数据
# ...
# 创建表格内容
table = TableContent(
headers=["日期", "产品", "数量", "金额"],
rows=[
["2023-01-01", "产品A", "10", "1000"],
["2023-01-02", "产品B", "5", "750"],
["2023-01-03", "产品A", "8", "800"]
]
)
return table
3. 富文本内容
from mcp.types import MarkdownContent, HTMLContent
@mcp.tool()
def generate_report(data: dict) -> MarkdownContent:
"""生成Markdown格式的报告"""
# 处理数据并生成报告
markdown = f"""
# {data['title']}
## 摘要
{data['summary']}
## 详细数据
- 项目1: {data['details']['item1']}
- 项目2: {data['details']['item2']}
"""
return MarkdownContent(markdown)
@mcp.tool()
def generate_html_report(data: dict) -> HTMLContent:
"""生成HTML格式的报告"""
html = f"""
<h1>{data['title']}</h1>
<h2>摘要</h2>
<p>{data['summary']}</p>
<h2>详细数据</h2>
<ul>
<li>项目1: {data['details']['item1']}</li>
<li>项目2: {data['details']['item2']}</li>
</ul>
"""
return HTMLContent(html)
4. 图像和二进制数据
from mcp.types import ImageContent, FileContent
@mcp.tool()
def generate_chart(data: List[float], title: str) -> ImageContent:
"""生成数据图表"""
import matplotlib.pyplot as plt
import io
import base64
# 创建图表
plt.figure()
plt.plot(data)
plt.title(title)
# 将图表保存为字节流
buf = io.BytesIO()
plt.savefig(buf, format='png')
buf.seek(0)
# 转换为base64编码
image_base64 = base64.b64encode(buf.read()).decode('utf-8')
# 返回图像内容
return ImageContent(
url=f"data:image/png;base64,{image_base64}",
alt_text=f"Chart: {title}"
)
@mcp.tool()
def generate_pdf_report(data: dict) -> FileContent:
"""生成PDF报告"""
# 生成PDF...
pdf_bytes = b"..." # 实际的PDF二进制数据
return FileContent(
filename="report.pdf",
content_type="application/pdf",
content=pdf_bytes
)
工具错误处理和异常传播
正确处理工具执行期间可能出现的错误是构建可靠MCP服务的关键。
1. 基本错误处理
最简单的方法是使用try/except捕获异常并返回错误信息：
@mcp.tool()
def divide(a: float, b: float) -> float:
"""除法计算"""
try:
return a / b
except ZeroDivisionError:
raise ValueError("除数不能为零")
except Exception as e:
raise ValueError(f"计算出错: {str(e)}")
2. 使用专用异常类
MCP提供了专用的异常类，可以提供更丰富的错误信息：
from mcp.errors import ToolExecutionError, ToolInputError
@mcp.tool()
def process_data(data: List[dict]) -> dict:
"""处理数据集合"""
try:
if not data:
raise ToolInputError("输入数据不能为空")
# 处理数据...
if some_processing_error:
raise ToolExecutionError(
"数据处理失败",
details={"error_item": problematic_item, "reason": "格式无效"}
)
return {"result": processed_data}
except (ToolInputError, ToolExecutionError):
# 直接让这些异常传播
raise
except Exception as e:
# 封装其他异常
raise ToolExecutionError(f"未预期的错误: {str(e)}")
3. 传递详细的错误状态
对于复杂工具，可能需要返回更详细的错误状态：
from mcp.types import ErrorContent
@mcp.tool()
def execute_operation(operation_type: str, parameters: dict) -> Union[dict, ErrorContent]:
"""执行特定操作"""
try:
# 尝试执行操作...
result = perform_operation(operation_type, parameters)
return result
except Exception as e:
# 返回结构化错误
return ErrorContent(
status_code=400 if isinstance(e, ValueError) else 500,
reason="Bad Request" if isinstance(e, ValueError) else "Internal Error",
detail=str(e),
help_text="请检查参数是否正确" if isinstance(e, ValueError) else "请联系管理员"
)
进阶示例：数据处理工具套件
下面是一个综合示例，展示如何创建一个数据处理工具套件：
import json
import pandas as pd
import matplotlib.pyplot as plt
import io
import base64
from typing import List, Dict, Any, Union, Optional
from mcp.server.fastmcp import FastMCP
from mcp.types import TableContent, ImageContent, ErrorContent
mcp = FastMCP("数据分析工具套件")
# 模拟数据存储
data_store = {}
@mcp.tool()
def load_csv_data(csv_content: str, dataset_name: str) -> Union[dict, ErrorContent]:
"""从CSV内容加载数据集
参数:
csv_content: CSV格式的数据内容
dataset_name: 数据集名称，用于后续引用
返回:
加载结果信息
"""
try:
# 使用StringIO处理CSV字符串
import io
# 解析CSV数据
df = pd.read_csv(io.StringIO(csv_content))
# 存储数据集
data_store[dataset_name] = df
return {
"status": "success",
"message": f"成功加载数据集 '{dataset_name}'",
"rows": len(df),
"columns": list(df.columns)
}
except Exception as e:
return ErrorContent(
status_code=400,
reason="Data Loading Error",
detail=str(e),
help_text="请检查CSV格式是否正确"
)
@mcp.tool()
def describe_dataset(dataset_name: str) -> Union[TableContent, ErrorContent]:
"""获取数据集的统计描述
参数:
dataset_name: 要描述的数据集名称
返回:
数据集的统计描述表格
"""
if dataset_name not in data_store:
return ErrorContent(
status_code=404,
reason="Dataset Not Found",
detail=f"找不到名为 '{dataset_name}' 的数据集",
help_text="请先使用load_csv_data加载数据集"
)
df = data_store[dataset_name]
description = df.describe().reset_index()
# 将DataFrame转换为表格格式
headers = description.columns.tolist()
rows = [row.tolist() for _, row in description.iterrows()]
# 将所有数据转换为字符串
rows = [[str(cell) for cell in row] for row in rows]
return TableContent(headers=headers, rows=rows)
@mcp.tool()
async def generate_chart(
dataset_name: str,
chart_type: str,
x_column: str,
y_column: str,
title: Optional[str] = None
) -> Union[ImageContent, ErrorContent]:
"""生成数据图表
参数:
dataset_name: 要使用的数据集名称
chart_type: 图表类型 (line, bar, scatter, pie)
x_column: X轴列名
y_column: Y轴列名
title: 图表标题（可选）
返回:
生成的图表图像
"""
# 检查数据集是否存在
if dataset_name not in data_store:
return ErrorContent(
status_code=404,
reason="Dataset Not Found",
detail=f"找不到名为 '{dataset_name}' 的数据集",
help_text="请先使用load_csv_data加载数据集"
)
df = data_store[dataset_name]
# 检查列是否存在
if x_column not in df.columns:
return ErrorContent(
status_code=400,
reason="Invalid Column",
detail=f"列 '{x_column}' 不存在",
help_text=f"可用列: {', '.join(df.columns)}"
)
if y_column not in df.columns:
return ErrorContent(
status_code=400,
reason="Invalid Column",
detail=f"列 '{y_column}' 不存在",
help_text=f"可用列: {', '.join(df.columns)}"
)
# 检查图表类型
valid_chart_types = ["line", "bar", "scatter", "pie"]
if chart_type not in valid_chart_types:
return ErrorContent(
status_code=400,
reason="Invalid Chart Type",
detail=f"图表类型 '{chart_type}' 无效",
help_text=f"有效的图表类型: {', '.join(valid_chart_types)}"
)
# 创建图表
plt.figure(figsize=(10, 6))
if chart_type == "line":
plt.plot(df[x_column], df[y_column])
elif chart_type == "bar":
plt.bar(df[x_column], df[y_column])
elif chart_type == "scatter":
plt.scatter(df[x_column], df[y_column])
elif chart_type == "pie":
# 饼图需要特殊处理
plt.pie(df[y_column], labels=df[x_column], autopct='%1.1f%%')
# 设置标题和标签
chart_title = title or f"{y_column} vs {x_column}"
plt.title(chart_title)
if chart_type != "pie":
plt.xlabel(x_column)
plt.ylabel(y_column)
plt.tight_layout()
# 保存图像到内存
buf = io.BytesIO()
plt.savefig(buf, format='png')
buf.seek(0)
# 转换为base64
image_base64 = base64.b64encode(buf.read()).decode('utf-8')
plt.close()
# 返回图像内容
return ImageContent(
url=f"data:image/png;base64,{image_base64}",
alt_text=chart_title
)
@mcp.tool()
def filter_data(
dataset_name: str,
filter_criteria: Dict[str, Any],
output_dataset_name: Optional[str] = None
) -> Union[dict, ErrorContent]:
"""根据条件筛选数据
参数:
dataset_name: 要筛选的数据集名称
filter_criteria: 筛选条件 {"column": value} 或 {"column": {"operator": ">=", "value": 100}}
output_dataset_name: 输出数据集名称，如果不提供则覆盖原数据集
返回:
筛选结果信息
"""
if dataset_name not in data_store:
return ErrorContent(
status_code=404,
reason="Dataset Not Found",
detail=f"找不到名为 '{dataset_name}' 的数据集"
)
df = data_store[dataset_name]
filtered_df = df.copy()
try:
# 应用筛选条件
for column, condition in filter_criteria.items():
if column not in df.columns:
return ErrorContent(
status_code=400,
reason="Invalid Column",
detail=f"列 '{column}' 不存在"
)
# 简单条件: {"column": value}
if not isinstance(condition, dict):
filtered_df = filtered_df[filtered_df[column] == condition]
continue
# 复杂条件: {"column": {"operator": ">=", "value": 100}}
operator = condition.get("operator")
value = condition.get("value")
if operator == "==":
filtered_df = filtered_df[filtered_df[column] == value]
elif operator == "!=":
filtered_df = filtered_df[filtered_df[column] != value]
elif operator == ">":
filtered_df = filtered_df[filtered_df[column] > value]
elif operator == ">=":
filtered_df = filtered_df[filtered_df[column] >= value]
elif operator == "<":
filtered_df = filtered_df[filtered_df[column] < value]
elif operator == "<=":
filtered_df = filtered_df[filtered_df[column] <= value]
elif operator == "in":
filtered_df = filtered_df[filtered_df[column].isin(value)]
elif operator == "not in":
filtered_df = filtered_df[~filtered_df[column].isin(value)]
elif operator == "contains":
filtered_df = filtered_df[filtered_df[column].str.contains(value, na=False)]
else:
return ErrorContent(
status_code=400,
reason="Invalid Operator",
detail=f"运算符 '{operator}' 无效",
help_text="有效的运算符: ==, !=, >, >=, <, <=, in, not in, contains"
)
# 保存结果
target_name = output_dataset_name or dataset_name
data_store[target_name] = filtered_df
return {
"status": "success",
"message": f"成功筛选数据集，结果保存为 '{target_name}'",
"original_rows": len(df),
"filtered_rows": len(filtered_df),
"filter_criteria": filter_criteria
}
except Exception as e:
return ErrorContent(
status_code=500,
reason="Filter Error",
detail=str(e)
)
@mcp.tool()
def aggregate_data(
dataset_name: str,
group_by_columns: List[str],
aggregations: Dict[str, str],
output_dataset_name: Optional[str] = None
) -> Union[dict, ErrorContent]:
"""聚合数据
参数:
dataset_name: 要聚合的数据集名称
group_by_columns: 分组列列表
aggregations: 聚合操作 {"column": "operation"}，operation可以是sum, avg, min, max, count
output_dataset_name: 输出数据集名称，如果不提供则生成新名称
返回:
聚合结果信息
"""
if dataset_name not in data_store:
return ErrorContent(
status_code=404,
reason="Dataset Not Found",
detail=f"找不到名为 '{dataset_name}' 的数据集"
)
df = data_store[dataset_name]
# 检查分组列
for col in group_by_columns:
if col not in df.columns:
return ErrorContent(
status_code=400,
reason="Invalid Column",
detail=f"分组列 '{col}' 不存在"
)
# 映射聚合操作
operation_map = {
"sum": "sum",
"avg": "mean",
"mean": "mean",
"min": "min",
"max": "max",
"count": "count"
}
agg_dict = {}
for col, op in aggregations.items():
if col not in df.columns:
return ErrorContent(
status_code=400,
reason="Invalid Column",
detail=f"聚合列 '{col}' 不存在"
)
if op.lower() not in operation_map:
return ErrorContent(
status_code=400,
reason="Invalid Operation",
detail=f"聚合操作 '{op}' 无效",
help_text=f"有效的聚合操作: {', '.join(operation_map.keys())}"
)
agg_dict[col] = operation_map[op.lower()]
try:
# 执行聚合
result_df = df.groupby(group_by_columns).agg(agg_dict).reset_index()
# 保存结果
target_name = output_dataset_name or f"{dataset_name}_aggregated"
data_store[target_name] = result_df
return {
"status": "success",
"message": f"成功聚合数据集，结果保存为 '{target_name}'",
"group_by": group_by_columns,
"aggregations": aggregations,
"result_rows": len(result_df),
"result_columns": list(result_df.columns)
}
except Exception as e:
return ErrorContent(
status_code=500,
reason="Aggregation Error",
detail=str(e)
)
if __name__ == "__main__":
mcp.run()
这个数据处理工具套件示例包含了：
- 数据加载工具
- 数据描述和统计工具
- 图表生成工具
- 数据筛选和聚合工具
- 错误处理和状态报告
它展示了如何设计一组相互配合的工具，以提供完整的数据处理功能。
小结
在本章中，我们深入探讨了MCP工具的高级用法：
- 工具设计的最佳实践
- 参数类型和验证
- 异步工具函数
- 返回复杂数据结构
- 错误处理和异常传播
- 通过数据处理工具套件，展示了复杂工具系统的实现
工具是MCP的核心功能，掌握这些高级用法将帮助您构建功能强大、易于使用的MCP服务。在下一章中，我们将探讨另一个重要概念：提示(Prompts)模板系统。