MCP 工具系统

工具（Tools）是 MCP 中让 AI 模型与外部系统交互的核心机制。工具设计为模型控制模式，即语言模型可以根据上下文理解和用户提示自动发现和调用工具。

工具基础

什么是工具？

工具是 AI 模型可以调用的函数。通过工具，LLM 可以：

• 执行计算
• 访问外部 API
• 读写文件
• 操作数据库
• 执行系统命令

工具结构

每个工具定义包含：

from mcp.types import Tool

Tool(
    name="tool_name",           # 唯一标识符
    description="工具描述",     # 说明工具用途
    inputSchema={              # 参数模式定义（JSON Schema）
        "type": "object",
        "properties": {
            "param_name": {
                "type": "string",
                "description": "参数描述"
            }
        },
        "required": ["param_name"]
    }
)

字段说明：

字段	类型	必需	说明
name	string	是	工具的唯一标识符
description	string	是	功能描述，帮助 LLM 理解何时使用
inputSchema	object	是	JSON Schema 格式的参数定义

定义工具

完整示例

创建一个天气查询工具：

from mcp.server import Server
from mcp.types import Tool, TextContent
import httpx

# 创建服务器实例
app = Server("weather-server")

# 1. 声明可用工具列表
@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="get_weather",
            description="获取指定城市的天气信息，包括温度、湿度、风力等",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，如：北京、上海"
                    },
                    "days": {
                        "type": "integer",
                        "description": "预报天数，1-7天",
                        "default": 1
                    }
                },
                "required": ["city"]
            }
        ),
        Tool(
            name="get_weather_by_coords",
            description="通过经纬度获取天气信息",
            inputSchema={
                "type": "object",
                "properties": {
                    "lat": {
                        "type": "number",
                        "description": "纬度"
                    },
                    "lon": {
                        "type": "number",
                        "description": "经度"
                    }
                },
                "required": ["lat", "lon"]
            }
        )
    ]

# 2. 处理工具调用
@app.call_tool()
async def call_tool(
    name: str, 
    arguments: dict | None
) -> list[TextContent]:
    if name == "get_weather":
        city = arguments.get("city")
        days = arguments.get("days", 1)
        
        # 调用天气 API
        weather_data = await fetch_weather(city, days)
        
        return [TextContent(
            type="text",
            text=format_weather(weather_data)
        )]
    
    elif name == "get_weather_by_coords":
        lat = arguments.get("lat")
        lon = arguments.get("lon")
        
        weather_data = await fetch_weather_by_coords(lat, lon)
        
        return [TextContent(
            type="text",
            text=format_weather(weather_data)
        )]
    
    raise ValueError(f"Unknown tool: {name}")

工具参数最佳实践

1. 提供清晰描述

# ❌ 差：描述不清
"city": {"type": "string"}

# ✅ 好：描述清晰
"city": {
    "type": "string", 
    "description": "城市名称，使用中文或英文，如：北京、Shanghai"
}

2. 设置默认值

"days": {
    "type": "integer",
    "description": "预报天数",
    "default": 1,  # 合理的默认值
    "minimum": 1,
    "maximum": 7
}

3. 使用枚举限制选项

"format": {
    "type": "string",
    "description": "输出格式",
    "enum": ["json", "xml", "text"]
}

4. 定义复杂参数

"filter": {
    "type": "object",
    "description": "过滤条件",
    "properties": {
        "status": {"type": "string"},
        "date_range": {
            "type": "object",
            "properties": {
                "start": {"type": "string", "format": "date"},
                "end": {"type": "string", "format": "date"}
            }
        }
    }
}

工具调用流程

完整流程图

协议消息详解

1. 列出工具 (tools/list)

// 请求
{
  "jsonrpc": "2.0",
  "id": "1",
  "method": "tools/list"
}

// 响应
{
  "jsonrpc": "2.0",
  "id": "1",
  "result": {
    "tools": [
      {
        "name": "get_weather",
        "description": "获取天气信息",
        "inputSchema": {...}
      }
    ]
  }
}

2. 调用工具 (tools/call)

// 请求
{
  "jsonrpc": "2.0",
  "id": "2",
  "method": "tools/call",
  "params": {
    "name": "get_weather",
    "arguments": {
      "city": "北京",
      "days": 1
    }
  }
}

// 响应
{
  "jsonrpc": "2.0",
  "id": "2",
  "result": {
    "content": [
      {
        "type": "text",
        "text": "北京今天天气晴，温度 15-25℃，微风"
      }
    ]
  }
}

3. 工具列表变更通知

如果服务器声明了 listChanged 能力，当工具列表变化时会发送通知：

{
  "jsonrpc": "2.0",
  "method": "notifications/tools/list_changed"
}

代码层面流程

# 1. 客户端获取工具列表
tools = await client.list_tools()

# 2. LLM 决定调用工具
# prompt: "北京今天天气怎么样？"
# LLM 决定调用 get_weather 工具，参数 city="北京"

# 3. 客户端执行工具调用
result = await client.call_tool(
    name="get_weather",
    arguments={"city": "北京", "days": 1}
)

# 4. LLM 使用结果生成响应
# "北京今天天气晴，温度 15-25℃，微风..."

工具返回类型

MCP 支持多种返回类型，工具结果可以包含多个不同类型的内容项：

1. 文本内容

from mcp.types import TextContent

return [TextContent(
    type="text",
    text="天气信息：晴，温度 15-25℃"
)]

2. 图像内容

from mcp.types import ImageContent

return [ImageContent(
    type="image",
    data="base64_encoded_image_data",
    mimeType="image/png"
)]

3. 嵌入资源

from mcp.types import EmbeddedResource, TextResourceContents

return [EmbeddedResource(
    type="resource",
    resource=TextResourceContents(
        uri="file:///data/report.pdf",
        mimeType="application/pdf",
        text="..."
    )
)]

4. 混合返回

# 返回多个内容项
return [
    TextContent(type="text", text="分析结果："),
    ImageContent(type="image", data="...", mimeType="image/png"),
    TextContent(type="text", text="详细说明...")
]

动态工具列表

服务器可以支持动态更新工具列表：

@app.list_tools()
async def list_tools() -> list[Tool]:
    # 每次都从最新状态生成工具列表
    tools = []
    
    # 基础工具
    tools.extend([
        Tool(name="echo", ...),
        Tool(name="calculate", ...)
    ])
    
    # 动态添加工具
    if self.is_weather_enabled:
        tools.append(Tool(name="get_weather", ...))
    
    if self.is_search_enabled:
        tools.append(Tool(name="search", ...))
    
    return tools

当工具列表变化时发送通知：

await session.send_notification(
    method="notifications/tools/list_changed"
)

错误处理

MCP 工具使用两种错误报告机制：

1. 协议错误

用于以下情况：

• 未知工具
• 无效参数
• 服务器错误

# 返回 JSON-RPC 错误
raise ToolNotFoundError(f"Unknown tool: {name}")
# 或
raise InvalidParamsError("Missing required parameter: city")

2. 工具执行错误

在工具结果中报告，设置 isError: true：

@app.call_tool()
async def call_tool(name: str, arguments: dict | None) -> list[TextContent]:
    try:
        if name == "get_weather":
            return await get_weather(arguments["city"])
        elif name == "search":
            return await search_web(arguments["query"])
    except ValidationError as e:
        # 参数验证错误
        return [TextContent(
            type="text",
            text=f"参数错误: {str(e)}"
        )]
    except APIError as e:
        # API 调用错误
        return [TextContent(
            type="text",
            text=f"API 错误: {str(e)}"
        )]
    except Exception as e:
        # 其他错误
        return [TextContent(
            type="text",
            text=f"发生错误: {str(e)}"
        )]

常见错误类型

错误类型	说明	处理方式
未知工具	工具名称不存在	返回协议错误
缺少参数	必需参数未提供	返回协议错误
参数类型错误	参数类型不匹配	返回协议错误
API 调用失败	外部服务不可用	返回工具执行错误
业务逻辑错误	如数据不存在	返回工具执行错误

安全考虑

服务器必须

• 验证所有工具输入
• 实施适当的访问控制
• 限制工具调用频率
• 清理工具输出

客户端应该

• 敏感操作需要用户确认
• 调用前向用户展示工具输入，避免恶意或意外数据泄露
• 传递给 LLM 前验证工具结果
• 为工具调用实现超时机制
• 记录工具使用情况用于审计

安全示例

# 敏感操作确认
@app.call_tool()
async def call_tool(name: str, arguments: dict | None):
    if name == "delete_file":
        # 要求用户确认
        if not await request_user_confirmation(
            f"确定要删除文件 {arguments['path']} 吗？"
        ):
            return [TextContent(
                type="text",
                text="操作已取消"
            )]
        
        # 执行删除
        return delete_file(arguments["path"])

工具设计原则

1. 单一职责：每个工具只做一件事
2. 清晰命名：名称清晰描述功能
3. 完整描述：description 要详细说明用途和使用场景
4. 参数验证：使用 JSON Schema 严格定义参数
5. 错误处理：提供有意义、用户友好的错误信息
6. 幂等设计：相同输入产生相同结果，便于重试

FastMCP 高级用法

Python SDK 提供了 FastMCP 类，使用装饰器语法简化工具定义。FastMCP 会自动从函数签名和文档字符串生成工具描述。

基础用法

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("my-server")

@mcp.tool()
def calculate(expression: str) -> str:
    """计算数学表达式
    
    Args:
        expression: 数学表达式，如 "2 + 3 * 4"
    
    Returns:
        计算结果
    """
    result = eval(expression)  # 注意：实际应用中应该使用安全的表达式解析
    return f"计算结果: {result}"

@mcp.tool()
async def fetch_url(url: str, timeout: int = 30) -> str:
    """获取网页内容
    
    Args:
        url: 要获取的网址
        timeout: 超时时间（秒），默认 30
    
    Returns:
        网页内容
    """
    import httpx
    async with httpx.AsyncClient() as client:
        response = await client.get(url, timeout=timeout)
        return response.text[:1000]  # 返回前 1000 字符

FastMCP 自动从：

• 函数参数类型推断 inputSchema
• 文档字符串提取参数描述
• 返回类型注解生成输出描述

运行服务器

if __name__ == "__main__":
    mcp.run()  # 默认使用 STDIO 传输

支持的类型

FastMCP 支持多种 Python 类型自动转换：

from typing import Literal
from pydantic import BaseModel, Field

class SearchParams(BaseModel):
    """搜索参数"""
    query: str = Field(description="搜索关键词")
    limit: int = Field(default=10, ge=1, le=100)

@mcp.tool()
def search(
    query: str,
    limit: int = 10,
    format: Literal["json", "text"] = "json"
) -> str:
    """搜索内容"""
    # ...
    return results

@mcp.tool()
def advanced_search(params: SearchParams) -> str:
    """高级搜索（使用 Pydantic 模型）"""
    # ...
    return results

能力声明

使用底层 Server 类时，需要手动声明 tools 能力：

from mcp.server import Server
from mcp.types import ServerCapabilities

app = Server("my-server", 
    options=ServerCapabilities(
        tools={"listChanged": True}
    )
)

使用 FastMCP 时，能力声明自动处理。

常见模式

文件操作工具

from pathlib import Path

@mcp.tool()
def read_file(path: str) -> str:
    """读取文件内容
    
    Args:
        path: 文件绝对路径
    
    Returns:
        文件内容
    """
    file_path = Path(path)
    if not file_path.exists():
        return f"错误: 文件 {path} 不存在"
    return file_path.read_text(encoding="utf-8")

@mcp.tool()
def list_directory(path: str) -> str:
    """列出目录内容
    
    Args:
        path: 目录路径
    
    Returns:
        目录中的文件和子目录列表
    """
    dir_path = Path(path)
    if not dir_path.is_dir():
        return f"错误: {path} 不是有效目录"
    
    items = []
    for item in dir_path.iterdir():
        item_type = "📁" if item.is_dir() else "📄"
        items.append(f"{item_type} {item.name}")
    
    return "\n".join(items)

API 调用工具

import httpx

@mcp.tool()
async def get_user(user_id: int) -> str:
    """获取用户信息
    
    Args:
        user_id: 用户 ID
    
    Returns:
        用户信息 JSON
    """
    async with httpx.AsyncClient() as client:
        response = await client.get(f"https://api.example.com/users/{user_id}")
        return response.json()

数据库查询工具

import sqlite3

@mcp.tool()
def query_database(sql: str) -> str:
    """执行 SQL 查询（只读）
    
    Args:
        sql: SELECT 查询语句
    
    Returns:
        查询结果
    """
    # 安全检查：只允许 SELECT
    if not sql.strip().upper().startswith("SELECT"):
        return "错误: 只允许执行 SELECT 查询"
    
    conn = sqlite3.connect("database.db")
    cursor = conn.execute(sql)
    columns = [desc[0] for desc in cursor.description]
    rows = cursor.fetchall()
    conn.close()
    
    # 格式化输出
    result = " | ".join(columns) + "\n"
    result += "-" * len(result) + "\n"
    for row in rows:
        result += " | ".join(str(v) for v in row) + "\n"
    
    return result

下一步

• 资源管理 - 学习如何提供上下文数据
• 提示模板 - 创建可重用的提示模板
• 采样机制 - 服务器发起的 LLM 调用

参考资料

• MCP 官方规范 - Tools
• MCP GitHub
• MCP Python SDK