本文基于 AgentScope 2.0.4 源码，示例项目为 myagent（AgentScope + FastAPI + Postgres + WebSocket）。

核心观点：工具不是 RPC，是 Agent 的能力扩展

工具不是 RPC，是 Agent 的能力扩展 – FunctionTool 包装的是函数，但执行权在 Agent 手里。

在第三篇文章中，你看到了 Agent 的 ReAct 循环：Agent 自己决定要不要调工具、调哪个工具、调几次。但那时我们没展开 – 工具到底是什么？怎么定义一个工具？Agent 怎么知道有哪些工具可用？

现在打开工具系统的黑盒。

你将掌握：工具定义、Toolkit、调用契约

• 理解工具的本质：工具不是 RPC，是 Agent 的能力扩展
• 掌握 FunctionTool 的定义（函数 + 自动 schema 推导）
• 理解 Toolkit 如何组装、工具如何分组
• 掌握 extra_agent_tools 注入（per-user 工具）
• 理解工具中间件（洋葱模型）
• 掌握工具调用契约（Toolkit.call_tool 内部）
• 理解工具结果（ToolResponse/ToolChunk/状态/截断）

前置知识：文章 1-3 与 async 概念

• 读过文章 1（Agent 认知基础）、文章 2（asyncio 基础）、文章 3（ReAct 循环）
• 特别理解文章 3 的 Acting 阶段（_batch_tool_calls / _acting_impl）
• 了解 Python 的 async 和生成器概念

一、工具的本质

工具不是 RPC

很多人把 Agent 的工具调用类比成 RPC（远程过程调用）——“Agent 调一个函数，跟调一个微服务差不多”。这个类比是错的，会误导你理解工具系统。

RPC 和 Agent 工具的根本区别在谁决定调用：

  ┌─ RPC ──────────────────────────────────────────────┐
  │                                                    │
  │  调用方 = 你的代码                                  │
  │  service.get_weather("北京")                       │
  │  # 你决定调哪个服务、传什么参数、怎么处理结果        │
  │  # 服务是被动的，等着被调                          │
  │                                                    │
  └────────────────────────────────────────────────────┘

  ┌─ Agent 工具 ───────────────────────────────────────┐
  │                                                    │
  │  调用方 = Agent（LLM）                              │
  │  Agent: "我需要查天气" -> 自己决定调 get_weather    │
  │  # Agent 决定调哪个工具、传什么参数、怎么用结果      │
  │  # 工具是被动的，但"是否被调"由 Agent 决定          │
  │                                                    │
  └────────────────────────────────────────────────────┘

RPC 是你编排调用；Agent 工具是 Agent 编排调用。工具定义者（你）只负责"提供能力 + 声明能力边界"，真正决定"何时用、怎么用"的是 Agent。

与 REST endpoint 的根本差异

  REST endpoint                       Agent FunctionTool
  ───────────────                      ────────────────
  路径 + HTTP 方法                     工具名 (name)
  @app.post("/weather")               FunctionTool(func=get_weather)
  请求体 -> 校验 -> 处理 -> 响应        LLM 生成参数 -> 执行 -> ToolChunk
  调用方 = 客户端 (你)                 调用方 = Agent (LLM)
  参数校验 = 手动 Pydantic            参数 schema 自动从函数签名推导
  返回 JSON                           返回 ToolChunk 流

REST endpoint 是"别人（客户端）来调用我"；FunctionTool 是"我注册一个能力，Agent 决定要不要用它"。

三类工具

AgentScope 的工具系统有三类（源码：agentscope/tool/__init__.py）：

  ToolBase (抽象基类)
    │
    ├── FunctionTool   包装一个 Python 函数（自定义工具的主力）
    │
    ├── MCPTool        MCP 协议工具（连外部 MCP server）
    │
    └── ToolGroup      工具分组（一个组内多个工具，可整体激活/停用）

• FunctionTool：把普通 Python 函数包装成 Agent 可调用的工具
• MCPTool：通过 MCP 协议连接外部工具服务（Model Context Protocol）
• ToolGroup：工具的分组容器，支持整体激活/停用

二、FunctionTool 详解

FunctionTool 是自定义工具的主力。它是把一个普通 Python 函数"翻译"成 Agent 能理解的工具（源码：agentscope/tool/_adapters.py）。

签名

FunctionTool(
    func,                 # 要包装的 Python 函数
    name=None,            # 工具名，默认用函数名
    description=None,     # 工具描述，默认从 docstring 提取
    is_concurrency_safe=True,  # 是否并发安全
    is_read_only=False,   # 是否只读
    is_state_injected=False,   # 是否注入 agent state
    middlewares=None,     # 工具中间件
)

自动 schema 推导

FunctionTool 的核心魔法是从函数签名自动推导参数 schema，不用你手写：

# 你只需要写一个普通 Python 函数
def calculate(expression: str) -> str:
    """计算数学表达式。"""
    return str(eval_expression(expression))

# FunctionTool 自动提取:
#   name = "calculate"
#   description = "计算数学表达式。"
#   input_schema = {"type":"object", "properties":{"expression":{"type":"string"}}, ...}
tool = FunctionTool(func=calculate)

这个 input_schema 会传给 LLM —— 告诉 LLM"这个工具接受一个叫 expression 的字符串参数"。LLM 生成工具调用时，就按这个 schema 生成 JSON 参数。

为什么 schema 重要

LLM 靠 schema 理解工具怎么用。如果 schema 不清晰，LLM 就会乱传参数。所以工具定义时：

• name 要语义化（如 get_weather，不要 func123）
• description 要说明用途和参数（LLM 靠这个决定"这个问题适不适合用这个工具"）
• 参数类型 要明确（str/int 还是 object）

返回值

FunctionTool 的函数返回 ToolChunk（增量）或 AsyncGenerator[ToolChunk]（流式）。AgentScope 的 Toolkit.call_tool 会累加这些 chunk 成完整的 ToolResponse。

三、Toolkit 组装

get_toolkit：工具的装配中心

get_toolkit() 是每次 chat turn 的工具装配中心（源码：agentscope/app/_service/_toolkit.py）。它把多个来源的工具组装成一个 Toolkit：

  get_toolkit() 的工具来源（按附加顺序）:
  ─────────────────────────────────────────────────────
  1. Workspace 内置      Bash / Read / Write / Grep / Glob / Edit
  2. Planning 工具        TaskCreate / TaskList / TaskGet / TaskUpdate
  3. 后台任务控制          ToolStop（来自 BackgroundTaskManager）
  4. 定时任务             ScheduleCreate / View / Delete / List
  5. 团队工具             TeamCreate / AgentCreate / TeamSay / TeamDelete
  6. 自定义扩展            extra_factory 提供的工具
  + workspace 的 skills 和 MCPs

每次 Agent 推理时，get_toolkit 把所有这些工具组装起来，传给 Agent 的 toolkit。

工具分组

Toolkit 用 ToolGroup 组织工具：

  Toolkit
    ├── ToolGroup("basic")     -> [Bash, Read, Write, ...]
    ├── ToolGroup("planning")  -> [TaskCreate, ...]
    ├── ToolGroup("schedule")  -> [ScheduleCreate, ...]
    └── ToolGroup("team")      -> [TeamCreate, ...]

工具组支持激活/停用（activate/deactivate）。停用的组里的工具，Agent 调不到。这提供了一层"按场景启用工具"的灵活性。

与 DI 容器的对比

AgentScope Toolkit	Spring DI 容器
工具注册	Bean 注册
ToolGroup 分组	按 profile 装配
工具激活/停用	@Profile / @Conditional
get_toolkit 组装	ApplicationContext 装配

四、extra_agent_tools

AgentToolFactory 签名

extra_agent_tools 是 create_app 的注入点，接收一个工厂函数：

# 签名: (user_id, agent_id, session_id) -> list[ToolBase]
async def create_tools(user_id, agent_id, session_id, **kwargs) -> list[ToolBase]:
    ...

框架在每次 chat turn 调用它，产出当前用户可用的工具列表。这就是"per-user 工具注入" —— 不同用户可能拿到不同的工具。

myagent 的 create_tools

# 源码: src/myagent/tools/registry.py
async def create_tools(user_id="", agent_id="", session_id="", **kwargs):
    _obs_mw = [ObservabilityMiddleware()]

    calculator_tool = TrustedFunctionTool(
        func=calculate,
        name="calculator",
        description="计算数学表达式...",
        middlewares=_obs_mw,
    )

    async def weather_handler(city=None):
        return await _get_weather(city, user_id=user_id, storage=storage)

    weather_tool = TrustedFunctionTool(
        func=weather_handler,
        name="get_weather",
        description="查询指定城市的当前天气...",
        middlewares=_obs_mw,
    )

    return [calculator_tool, weather_tool]

注意 weather_handler 闭包捕获了 user_id —— 这样天气工具就能根据当前用户自动定位城市。这就是 per-user 注入的价值：工具能用上当前请求的上下文。

五、工具中间件

ToolMiddlewareBase

工具中间件基于 ToolMiddlewareBase，用洋葱模型包裹工具调用（源码：agentscope/tool/_base.py）。它和文章 3 讲过的 Agent 中间件不同 —— 工具中间件专门包裹单个工具的调用。

myagent 的 ObservabilityMiddleware

# 源码: src/myagent/observability.py
class ObservabilityMiddleware(ToolMiddlewareBase):
    async def on_tool_call(self, tool, input_kwargs, next_handler):
        logger.info("tool_call_start name=%s params=%s", tool.name, input_kwargs)
        start = time.monotonic()
        try:
            async for chunk in next_handler(**input_kwargs):
                yield chunk
        except Exception as e:
            logger.error("tool_call_error name=%s error=%s", tool.name, e)
            raise
        else:
            logger.info("tool_call_end name=%s duration_ms=%.1f",
                        tool.name, (time.monotonic()-start)*1000)

这个中间件在每个工具调用前后打日志：开始、结束（含耗时）、出错。这就是可观测性 —— 你在外面包一层，不用改工具本身。

TrustedFunctionTool 的权限检查

工具执行前会调用 check_permissions() 决定是否允许。默认的 FunctionTool.check_permissions() 返回 ASK（需要用户确认）：

# FunctionTool 默认: 需要用户确认
async def check_permissions(self, *_args, **_kwargs):
    return PermissionDecision(
        behavior=PermissionBehavior.ASK,
        message="Custom function tools must be explicitly allowed.",
    )

myagent 继承了 FunctionTool 并重写为自动放行：

# 源码: src/myagent/tools/registry.py
class TrustedFunctionTool(FunctionTool):
    async def check_permissions(self, *_args, **_kwargs):
        return PermissionDecision(
            behavior=PermissionBehavior.ALLOW,
            message="Trusted tool auto-allowed.",
        )

这个设计很关键：如果自定义工具都要求用户确认，聊天会频繁卡在权限确认上（死锁）。myagent 用 TrustedFunctionTool 自动放行自定义工具，避免死锁。关于权限系统的完整解析留到文章 13。

六、工具调用契约

现在打开 Toolkit.call_tool 内部 —— 这是"单个工具如何被调用"的完整契约。注意文章 3 讲的是"循环如何调度工具"（_batch_tool_calls / _acting_impl），这里讲的是"一个工具被调用时，内部发生了什么"。

# 源码: agentscope/tool/_toolkit.py Toolkit.call_tool
async def call_tool(self, tool_call, state):
    tool_response = ToolResponse(id=tool_call.id)

    # 1. 工具存在性检查
    available_tools = await self._get_available_tools(...)
    if tool_call.name not in available_tools:
        chunk = ToolChunk(content=[TextBlock(text=f"ToolNotFoundError: ...")],
                          state=ToolResultState.ERROR)
        yield chunk
        return

    # 2. 工具组激活检查
    #    工具在未激活的 group -> ToolGroupInactiveError

    # 3. 参数解析
    kwargs = _json_loads_with_repair(tool_call.input)
    #    LLM 生成的参数 JSON 可能不合法，需要修复解析

    # 4. state 注入
    if tool_func.is_state_injected:
        kwargs["_agent_state"] = state

    # 5. 返回值分派
    if inspect.iscoroutinefunction(tool_func.__call__):
        res = await tool_func(**kwargs)
    else:
        res = tool_func(**kwargs)

    if isinstance(res, ToolChunk):
        yield res
    elif isinstance(res, AsyncGenerator):
        async for chunk in res:
            yield chunk
    elif isinstance(res, Generator):
        for chunk in res:
            yield chunk

    # 6. 异常处理
    #    McpError / DeveloperOrientedException
    #    -> 包装成 ToolChunk(state=ERROR)

六个步骤

1. 存在性检查：工具名在不在可用列表里。不在 -> 返回 ToolNotFoundError 的 ToolChunk（不让异常向上传播，而是给 Agent 一个错误反馈，让 Agent 决定怎么办）。
2. 工具组激活检查：工具在未激活的组里 -> ToolGroupInactiveError，提示 Agent 先激活组。
3. 参数解析：_json_loads_with_repair 解析 LLM 生成的参数 JSON。LLM 生成的 JSON 可能带多余字符、格式不严，需要"修复式解析"。
4. state 注入：如果工具声明了 is_state_injected，把 agent state 注入参数，工具可以读写 agent 状态。
5. 返回值分派：根据工具函数类型（同步/异步/生成器/异步生成器）分派执行，逐个 yield ToolChunk。
6. 异常处理：工具函数执行时若抛异常，call_tool 会捕获并包装成 ERROR 状态的 ToolChunk，给 Agent 反馈而不是让异常向上传播（DeveloperOrientedException 这类开发者异常除外，会重新抛出）。

关键洞察

工具调用契约的本质：工具系统的设计原则是把执行异常转换为错误状态的 ToolChunk 反馈给 Agent，而不是让异常向上传播导致流程崩溃。

  传统函数: 出错了 -> raise Exception -> 调用方 try/except
  Agent 工具: 出错了 -> call_tool 捕获 -> 返回 ToolChunk(state=ERROR)
                       -> Agent 自己决定怎么处理

这里有个细微差别：工具函数内部可以抛异常，但 Toolkit.call_tool 会拦截并把它包装成 ERROR 状态的 ToolChunk。所以从 Agent 的视角看，它"不会收到异常"，只会看到带错误信息的 ToolChunk。唯一的例外是 DeveloperOrientedException（框架开发者错误），它会被重新抛出。Agent 不能像传统代码那样 try/except —— 它是 LLM，需要"看到错误信息然后决定下一步"。所以工具系统把错误转成 ToolChunk 反馈给 Agent，让 Agent 自主决策。

七、工具结果

ToolResponse / ToolChunk

  ToolChunk      增量结果（流式，工具执行过程中逐步产生）
  ToolResponse   完整结果（Toolkit.call_tool 累加所有 chunk 后生成）

工具可以返回单个 ToolChunk，也可以 yield 一串 ToolChunk（流式）。Toolkit.call_tool 内部会累加这些 chunk 到 ToolResponse。

ToolResultState

工具结果有状态（源码：agentscope/message）：

  ToolResultState:
    SUCCESS     成功
    ERROR       出错（工具异常/参数错误）
    INTERRUPTED 被中断（文章 3 讲过 _close_unfinished_tool_calls 会标记 INTERRUPTED）

结果截断

工具结果可能很大（比如读了一个大文件）。AgentScope 的 ContextConfig.tool_result_limit（默认 50000 token）控制工具结果的最大长度，超出的截断，防止撑爆 agent context。

八、myagent 实战

回顾 myagent 的完整工具链路（源码：src/myagent/tools/）：

  server.py
    create_app(extra_agent_tools=create_tools, ...)   # 注入工具工厂
      └─ 每次 chat turn
         └─ get_toolkit(extra_factory=create_tools)   # 组装工具
            └─ create_tools(user_id, agent_id, session_id)
               ├─ TrustedFunctionTool(calculate, "calculator")
               └─ TrustedFunctionTool(weather_handler, "get_weather")
                   └─ 两个工具都带 ObservabilityMiddleware

• calculator：calculate 函数，用 ast 安全解析数学表达式（源码：tools/calculator.py）
• get_weather：weather_handler 闭包，捕获 user_id，调用高德天气 API（源码：tools/weather.py）
• TrustedFunctionTool：重写 check_permissions 返回 ALLOW，避免权限确认死锁
• ObservabilityMiddleware：包裹两个工具，记录调用耗时和状态

九、常见陷阱

陷阱 1: 工具 description 不清晰导致 LLM 误调

症状：LLM 不调某个工具，或乱传参数。

原因：工具的 description 和参数 schema 不清晰。LLM 靠这些理解"什么时候该用这个工具、参数怎么传"。

解决：description 写清楚用途 + 参数说明。比如：

FunctionTool(
    func=calculate,
    name="calculator",
    description="计算数学表达式，支持加减乘除、幂、取模。参数 expression: 数学表达式字符串",
)

而不是 description="工具" 这种模糊描述。

陷阱 2: 同步工具阻塞事件循环

症状：Agent 调用工具时，所有用户请求卡住。

原因：工具函数是同步阻塞的（如 requests.get、time.sleep），在 async 环境里阻塞了事件循环。

解决：工具函数用异步实现（httpx.AsyncClient），或用 asyncio.to_thread 跑同步代码。参考文章 2 的"同步阻塞混入 async"陷阱。

陷阱 3: 工具结果过大撑爆 context

症状：Agent 对话越来越慢，甚至报 context 超长错误。

原因：工具返回了超大结果（读大文件、查大量数据），撑爆了 agent context。

解决：设置 ContextConfig.tool_result_limit 限制工具结果长度；工具函数自身也尽量只返回摘要而非全量数据。

总结：工具定义了什么能做，Agent 决定做什么

工具系统是 Agent 的能力扩展。它不是 RPC —— 是"你注册能力，Agent 决定用不用"。核心组件：

• FunctionTool：把 Python 函数包装成工具，自动推导 schema
• Toolkit：组装多来源工具（workspace/planning/schedule/team/extras/skills/mcp）
• extra_agent_tools：per-user 工具注入
• 工具中间件：洋葱模型包裹工具调用（观测/权限）
• 工具调用契约：捕获执行异常为错误 ToolChunk 让 Agent 决策
• 工具结果：ToolChunk 流 + ToolResponse + 状态 + 截断

理解工具系统，你就理解了 Agent 能力的边界 —— 工具定义了什么能做，Agent 决定做什么。

下一篇文章将深入事件流模型 —— Agent 怎么用事件思考和沟通。你已经理解了工具如何被调用，接下来看这些调用怎么变成前端可见的事件流。

进一步探索：并发安全、权限恢复、超大结果

• 如果一个工具是纯计算（calculator），和查数据库的工具，它们的 is_read_only / is_concurrency_safe 应该怎么设？
• check_permissions 返回 ASK 时，前端要展示什么？Agent 暂停后怎么恢复？
• 工具返回超大结果时，除了截断，还有什么策略？（如分层摘要、只返回链接）