mcpField
A
anthropic-tool-mastery

anthropic-tool-mastery

github.com/itallstartedwithaideaAgent 工作流4475 次安装28 周前更新
代码仓库

关于 anthropic-tool-mastery

Anthropic Tool Mastery 是用于设计和协调 Agent 中 Claude 工具调用的技能,涵盖工具 schema、并行调用、流式结果、扩展思考、结果组合、错误恢复与缓存,也包含适用于自定义 MCP 服务工具的模式。

主要功能

  1. 01设计工具 schema
  2. 02并行分派工具调用
  3. 03流式传输工具结果
  4. 04使用扩展思考规划
  5. 05处理部分失败并组合结果
  6. 06缓存工具结果

适用场景

  1. 01构建单轮调用多个工具的 Agent
  2. 02提升工具选择和参数准确性
  3. 03协调多工具工作流
  4. 04为长时间运行的工具提供进度

MCP 相关

介绍 Claude 使用自定义 MCP 服务工具的方法,包括工具定义、并行分派和结果处理模式;项目本身并非 MCP 服务。

安装

复制下面的命令,在终端中运行。

$npx skills add https://github.com/itallstartedwithaidea/agent-skills --skill anthropic-tool-mastery

安装后在 Claude Code 中输入/anthropic-tool-mastery确认已加载。

详细介绍

Anthropic 工具精通

Agent Skills™ 的一部分,由 googleadsagent.ai™ 提供

描述

Anthropic 工具精通是充分发挥 Claude 原生 tool_use 能力的权威指南。Claude 的工具系统不仅仅是函数调用接口,更是一种结构化推理协议,能够让模型拆解问题、并行调度操作、处理流式结果,并通过扩展思考串联工具输出。掌握这些模式,才能让 Agent 从一次笨拙地调用一个工具,进阶为像资深工程师一样流畅地编排复杂的多工具工作流。

本技能总结了 Buddy™ Agent 在 googleadsagent.ai™ 中验证有效的模式。在该 Agent 中,工具编排会在单个推理轮次内处理并发的 Google Ads API 调用、网页搜索、文件操作和分析计算。相关技巧涵盖工具定义设计(架构质量会直接影响调用准确性)、并行调度(在单个轮次中发起多个独立的工具调用)、结果组合(合并并行调用的输出)以及错误恢复(处理多工具批次中的部分失败)。

高级模式包括流式传输工具结果以提供实时反馈、集成扩展思考(在执行前使用思考块规划工具调用序列),以及缓存工具结果以避免重复调用。这些技巧可直接应用于 Claude Code 的内置工具,也适用于自定义 MCP 服务工具。

适用场景

  • 构建需要在每个推理轮次调用多个工具的 Agent
  • 工具调用准确率低于可接受的水平(参数错误、选错工具)
  • 需要获取长时间运行的工具操作的实时流式反馈
  • 需要使用扩展思考来辅助工具选择和参数构建
  • 多工具工作流需要协调和结果组合
  • 需要设计与模型兼容性最高的自定义工具

工作原理

sequenceDiagram
    participant U as User/System
    participant C as Claude (Extended Thinking)
    participant T1 as Tool A
    participant T2 as Tool B
    participant T3 as Tool C

    U->>C: Task + Tool Definitions
    Note over C: Extended Thinking:<br/>Plan tool sequence
    C->>T1: Parallel Call 1
    C->>T2: Parallel Call 2
    T1-->>C: Result A (streamed)
    T2-->>C: Result B (streamed)
    Note over C: Extended Thinking:<br/>Compose results, plan next step
    C->>T3: Sequential Call (depends on A+B)
    T3-->>C: Result C
    C->>U: Final composed response

Claude 的工具执行模型支持在单个助手轮次内并行调度彼此独立的工具调用。模型会使用扩展思考来规划工具调用序列,识别哪些调用彼此独立(可以并行执行),哪些调用存在依赖关系(必须按顺序执行)。流式传输可以在长时间运行的工具执行期间提供实时进度反馈。所有工具结果返回后,扩展思考会将结果整合为连贯的响应,或规划下一轮工具调用。

实现

最佳工具定义模式:

{
  "name": "search_knowledge_base",
  "description": "Search the domain knowledge base for patterns matching a query. Returns ranked results with confidence scores. Use this BEFORE generating recommendations to ground them in verified patterns.",
  "input_schema": {
    "type": "object",
    "properties": {
      "query": {
        "type": "string",
        "description": "Natural language search query describing the pattern or information needed"
      },
      "category": {
        "type": "string",
        "enum": ["bidding", "targeting", "creative", "budget", "general"],
        "description": "Knowledge category to search within"
      },
      "max_results": {
        "type": "integer",
        "default": 5,
        "description": "Maximum number of results to return (1-20)"
      }
    },
    "required": ["query"]
  }
}

并行工具调用处理程序:

interface ToolCall {
  id: string;
  name: string;
  input: Record<string, unknown>;
}

interface ToolResult {
  tool_use_id: string;
  content: string;
  is_error: boolean;
}

async function executeToolCalls(calls: ToolCall[], registry: ToolRegistry): Promise<ToolResult[]> {
  const results = await Promise.allSettled(
    calls.map(async (call) => {
      const handler = registry.get(call.name);
      if (!handler) {
        return { tool_use_id: call.id, content: `Unknown tool: ${call.name}`, is_error: true };
      }
      try {
        const result = await handler.execute(call.input);
        return { tool_use_id: call.id, content: JSON.stringify(result), is_error: false };
      } catch (error) {
        return { tool_use_id: call.id, content: `Error: ${String(error)}`, is_error: true };
      }
    })
  );

  return results.map((r) =>
    r.status === "fulfilled" ? r.value : { tool_use_id: "", content: `Execution failed: ${r.reason}`, is_error: true }
  );
}

流式工具结果集成:

async def stream_tool_execution(client, messages, tools):
    """Execute tools with streaming for real-time progress."""
    async with client.messages.stream(
        model="claude-sonnet-4-20250514",
        max_tokens=8096,
        tools=tools,
        messages=messages,
    ) as stream:
        tool_calls = []
        async for event in stream:
            if event.type == "content_block_start" and event.content_block.type == "tool_use":
                tool_calls.append({
                    "id": event.content_block.id,
                    "name": event.content_block.name,
                    "input_json": "",
                })
            elif event.type == "content_block_delta" and hasattr(event.delta, "partial_json"):
                tool_calls[-1]["input_json"] += event.delta.partial_json
            elif event.type == "content_block_stop" and tool_calls:
                call = tool_calls[-1]
                call["input"] = json.loads(call["input_json"])

        if tool_calls:
            results = await execute_parallel_tools(tool_calls)
            messages.append({"role": "assistant", "content": stream.get_final_message().content})
            messages.append({"role": "user", "content": results})
            return await stream_tool_execution(client, messages, tools)

        return stream.get_final_message()

工具结果缓存:

class ToolCache:
    def __init__(self, ttl_seconds=300):
        self.cache = {}
        self.ttl = ttl_seconds

    def cache_key(self, tool_name: str, params: dict) -> str:
        normalized = json.dumps(params, sort_keys=True)
        return f"{tool_name}:{hashlib.sha256(normalized.encode()).hexdigest()[:16]}"

    async def execute_with_cache(self, tool_name, params, executor):
        key = self.cache_key(tool_name, params)
        if key in self.cache:
            entry = self.cache[key]
            if time.time() - entry["timestamp"] < self.ttl:
                return entry["result"]

        result = await executor(tool_name, params)
        self.cache[key] = {"result": result, "timestamp": time.time()}
        return result

最佳实践

  1. 编写能说明工具适用时机的描述——模型会根据描述选择工具;“在生成建议之前使用此工具”比“搜索数据库”更有效。
  2. 对受限参数使用枚举——参数的有效值固定时,务必使用枚举;这样可以避免一整类参数错误。
  3. 为并行调度而设计——尽可能让工具彼此独立且无副作用;模型会自然地并行执行独立调用。
  4. 妥善处理部分失败——4 个并行工具调用中有 3 个成功时,应使用这 3 个成功的结果,而不是让整个批次失败。
  5. 缓存确定性工具结果——如果对相同数据执行相同查询会得到相同结果,就应缓存该结果,以避免重复计算和消耗 token。
  6. 精简工具结果负载——大型工具结果会占用上下文窗口;只返回模型需要的字段,而非整行数据库记录。
  7. 对工具架构进行版本管理——工具接口变更时,应更新描述,引导模型使用新的参数格式。

平台兼容性

功能Claude CodeCursorCodexGemini CLI
原生 tool_use✅ 完整支持✅ 通过 API⚠️ 函数调用⚠️ 函数调用
并行调度✅ 原生支持✅ 原生支持✅ 支持✅ 支持
流式结果✅ 完整支持✅ 完整支持✅ 支持✅ 支持
扩展思考✅ 原生支持⚠️ 取决于模型❌ 不可用⚠️ 有限支持
工具结果缓存✅ 自定义✅ 自定义✅ 自定义✅ 自定义

相关技能

  • 提示词架构 - 结构化提示词设计,可优化工具选择准确性和参数构建
  • 上下文工程 - Token 预算管理,确保工具结果适配上下文限制
  • 多 Harness 可移植性 - 跨平台工具抽象,用于编写可在所有 Harness 中运行的工具集成

关键词

tool-use, parallel-tool-calling, streaming, extended-thinking, tool-definitions, tool-orchestration, result-composition, error-recovery, tool-caching, agent-skills


© 2026 googleadsagent.ai™ | Agent Skills™ | MIT License

评论

支持 Markdown