AI 智能体博客
计算机访问系统

工具使用模式：构建可靠的智能体-工具接口

agentsarchitecturetutorialclaudemcp

你的智能体调用了一个工具，收到了40行JSON数据——原始API响应，嵌套对象，错误代码埋在status字段里。模型读取了它，挑了一个看起来合理的值，然后继续。这个值是错的。三步之后，智能体自信地写了一份基于错误数据的报告。

工具运行正常。接口失败了。

工具使用是将语言模型转变为智能体的机制。智能体的每项能力——搜索数据库、写入文件、调用API、查询服务——都通过工具接口到达。如果接口设计不当，即使底层服务运行正常，模型也会做出更差的决策。本指南涵盖构建精确、可靠、生产就绪的工具接口的五种模式。

前提条件： 熟悉Python和Claude API。关于MCP作为工具传输层的背景，参见构建你的第一个MCP服务器。

为什么接口设计很重要

当智能体选择并使用工具时，它做出两个决策：

  1. 调用哪个工具 — 由工具的name和description驱动
  2. 传递什么参数 — 由工具的input_schema驱动

模糊的描述导致错误的工具选择。宽松的模式允许模型传递格式错误的输入。非结构化的结果让模型猜测发生了什么。大多数智能体错误不在推理中——而在工具边界处。

模式1：模式优先设计

在编写实现之前先写JSON模式。严格的模式在输入阶段约束模型行为——在任何东西执行之前。

import anthropic


client = anthropic.Anthropic()


tools = [
    {
        "name": "search_products",
        "description": (
            "按关键词搜索产品目录。"
            "返回带有ID、名称和价格的匹配产品列表。"
            "当用户想要查找或浏览产品时使用。"
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "搜索关键词"
                },
                "category": {
                    "type": "string",
                    "enum": ["electronics", "clothing", "food", "home", "all"],
                    "description": "要筛选的产品类别。未指定时使用'all'。"
                },
                "max_results": {
                    "type": "integer",
                    "minimum": 1,
                    "maximum": 20,
                    "description": "返回结果数量。默认：5"
                }
            },
            "required": ["query", "category"]
        }
    }
]


response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    tools=tools,
    messages=[{"role": "user", "content": "找100元以下的电子产品"}]
)

减少错误的模式规则：

  • 对具有固定有效值集合的任何字段使用enum
  • 在数值字段上设置minimum/maximum以防止超出范围的输入
  • 仅在工具真的无法在没有该字段时运行的情况下才将字段标记为required
  • 从模型的角度编写描述：“当……时使用”告诉模型何时调用工具

何时使用： 每个工具定义。

模式2：结构化工具结果

返回类型化的、机器可读的结果。永远不要返回原始API响应或散文描述。

import json
from dataclasses import dataclass, asdict
from typing import Any, Optional




@dataclass
class ToolResult:
    success: bool
    data: Optional[Any] = None
    error: Optional[str] = None


    def to_content(self) -> str:
        return json.dumps(asdict(self), ensure_ascii=False)




def search_products(
    query: str,
    category: str,
    max_results: int = 5,
) -> ToolResult:
    try:
        raw_results = _query_database(query, category, limit=max_results)
        products = [
            {"id": r["product_id"], "name": r["title"], "price": r["price_usd"]}
            for r in raw_results
        ]
        return ToolResult(success=True, data={"products": products, "count": len(products)})
    except ConnectionError as e:
        return ToolResult(success=False, error=f"数据库不可用：{e}")
    except Exception as e:
        return ToolResult(success=False, error=f"搜索失败：{type(e).__name__}: {e}")




def _query_database(query, category, limit):
    return []

一致的{success, data, error}信封意味着模型始终知道在哪里查找。关于优雅处理故障，参见智能体错误恢复模式。

模式3：并行工具调用

Claude可以在单个响应中请求多个工具。并行处理它们而不是顺序处理。

from concurrent.futures import ThreadPoolExecutor, as_completed




TOOL_REGISTRY = {
    "search_products": search_products,
}




def dispatch_tool(name: str, inputs: dict) -> ToolResult:
    handler = TOOL_REGISTRY.get(name)
    if not handler:
        return ToolResult(success=False, error=f"未知工具：{name}")
    return handler(**inputs)




def process_tool_calls(response: anthropic.types.Message) -> list[dict]:
    tool_uses = [
        block for block in response.content
        if block.type == "tool_use"
    ]
    if not tool_uses:
        return []


    def execute(tool_use):
        result = dispatch_tool(tool_use.name, tool_use.input)
        return {
            "type": "tool_result",
            "tool_use_id": tool_use.id,
            "content": result.to_content(),
        }


    with ThreadPoolExecutor(max_workers=len(tool_uses)) as executor:
        futures = {executor.submit(execute, tu): tu for tu in tool_uses}
        results = []
        for future in as_completed(futures):
            results.append(future.result())


    return results




def run_agent(user_message: str) -> str:
    messages = [{"role": "user", "content": user_message}]


    while True:
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=4096,
            tools=tools,
            messages=messages,
        )


        if response.stop_reason == "end_turn":
            for block in response.content:
                if hasattr(block, "text"):
                    return block.text
            return ""


        if response.stop_reason == "tool_use":
            tool_results = process_tool_calls(response)
            messages.append({"role": "assistant", "content": response.content})
            messages.append({"role": "user", "content": tool_results})
        else:
            break


    return ""

关于编排每个调用工具的多个智能体，参见多智能体模式。

模式4：安全工具调用包装器

永远不要让工具异常在未处理的情况下到达智能体循环。

import signal




def timeout_handler(signum, frame):
    raise TimeoutError("工具执行超时")




def safe_tool_call(
    name: str,
    inputs: dict,
    timeout_seconds: int = 30,
) -> ToolResult:
    """
    带超时执行工具，捕获所有异常。
    始终返回ToolResult——从不引发异常。
    """
    signal.signal(signal.SIGALRM, timeout_handler)
    signal.alarm(timeout_seconds)
    try:
        return dispatch_tool(name, inputs)
    except TimeoutError:
        return ToolResult(
            success=False,
            error=f"工具'{name}'在{timeout_seconds}秒后超时"
        )
    except Exception as e:
        return ToolResult(
            success=False,
            error=f"工具'{name}'引发{type(e).__name__}: {e}"
        )
    finally:
        signal.alarm(0)

当工具返回success: false时，模型可以决定是重试、尝试替代方案还是向用户报告失败。关于更广泛的重试策略，参见错误恢复模式。

模式5：结果验证和截断

在将工具结果返回给模型之前进行验证。

MAX_TOOL_RESULT_CHARS = 8000




def validate_result(result: ToolResult, expected_keys: list[str]) -> ToolResult:
    if not result.success or not isinstance(result.data, dict):
        return result
    missing = [k for k in expected_keys if k not in result.data]
    if missing:
        return ToolResult(
            success=False,
            error=f"工具响应缺少预期字段：{missing}"
        )
    return result




def truncate_result(result: ToolResult) -> ToolResult:
    content = result.to_content()
    if len(content) <= MAX_TOOL_RESULT_CHARS:
        return result


    truncated_data = {
        "truncated": True,
        "chars_omitted": len(content) - MAX_TOOL_RESULT_CHARS,
        "content": content[:MAX_TOOL_RESULT_CHARS],
    }
    return ToolResult(
        success=result.success,
        data=truncated_data,
        error="结果已截断——对于上下文窗口来说太大了",
    )




def safe_tool_call_validated(
    name: str,
    inputs: dict,
    expected_keys: list[str] | None = None,
) -> ToolResult:
    result = safe_tool_call(name, inputs)
    if expected_keys:
        result = validate_result(result, expected_keys)
    result = truncate_result(result)
    return result

关于在生产中观察和调试这些失败，参见调试和可观察性。

常见错误

错误1：返回原始API响应

模型收到一个有30个字段的嵌套对象，大多数不相关。它选择了错误的那个。

解决方案： 在返回之前格式化响应。只返回模型做下一个决策所需的内容。

错误2：有副作用但无确认的工具

发送电子邮件、删除记录或扣款的工具不应该静默执行。

解决方案： 对于不可逆的操作，使用双工具模式：plan_email返回预览，send_email实际发送。

错误3：工具职责重叠

两个做类似事情的工具迫使模型猜测该用哪个。

解决方案： 每个工具应该有独特的、不重叠的目的。

错误4：外部工具没有超时

一个缓慢的第三方API调用会无限期地阻塞整个智能体循环。

解决方案： 始终设置超时（模式4）。

生产检查清单

  • 每个工具都有从模型角度写的描述（“当……时使用”）
  • 所有固定值输入使用enum字段
  • 每个工具返回{success, data, error}——从不返回原始响应
  • 工具调用包装在safe_tool_call中
  • 多工具响应的并行执行
  • 每个外部工具设置超时
  • 可变大小负载的结果截断
  • 调用外部API的工具进行验证

下一步

  1. 从模式开始 — 在函数体之前编写input_schema
  2. 为每个现有工具添加ToolResult包装器
  3. 引入safe_tool_call 来加固你的智能体循环
  4. 为调用你不控制的API的工具设置结果验证

相关指南：