agentscope 源码分析（1）：Toolkit 类实现

AgentScope 是一个面向生产环境、易于使用的智能体框架，提供灵活的核心抽象以适配日益增强的大模型能力，并内置微调支持。其专为日益具备自主行为能力的大语言模型（agentic LLM）而设计，充分利用模型自身的推理与工具调用能力，而非通过僵化的提示词和预设流程对其加以限制。

本系列文章将会详细分析 AgentScope 的源码实现，这篇文章将先从其 工具调用 功能的实现开始。

工具调用的基础知识

如果说大模型（LLM）是负责思考和决策的 大脑，那么工具就是 Agent 的 五官 与 肢体。正是因为有了 工具调用，AI 才完成了从 聊天机器人 到 智能体 的本质蜕变。在 AI Agent 框架中，如何高效、精准地实现 工具调用（Tool/Function Calling）是构建智能体的工程核心。

工具调用 的实现本质上是一个 模型 与 系统 之间的多轮会话协议，通常遵循以下四个步骤：

• 注册与定义 (Registration)：开发者预先定义好工具的 说明书。这通常是一个 JSON Schema，包含：

  • 函数名：如 get_weather。
  • 描述：解释这个函数能做什么（LLM 靠这个理解何时调用）。
  • 参数结构：定义参数类型、名称及是否必填

• 模型决策 (Decision)：当用户输入请求时，LLM 会同时接收 用户问题 和 工具定义

  • 语义匹配：LLM 发现用户问题（北京天气怎么样）与某个工具的描述匹配
  • 输出格式化：LLM 停止生成自然语言，转而输出一段符合 JSON 格式的文本，其中包含它认为需要的参数

• 外部执行 (Execution)：这是关键点，LLM 本身不运行代码

  • Agent 应用程序解析 LLM 输出的 JSON
  • 程序代码根据解析出的函数名和参数，真实地调用外部 API 或数据库
  • 获取返回结果

• 结果注入：程序将外部工具的返回结果拼接回对话历史中，再次发送给 LLM

  • LLM 看到：用户问了天气 -> 我要求调用函数 -> 函数返回了 15°C
  • LLM 最终生成人类可读的回复：上海现在的天气晴朗，气温 15°C

一个简单示例

下面通过一个极简的例子，展示了 LLM 工具调用的核心原理。各个 Agent 框架在实现工具调用时，本质上就是通过各种工程技巧，简化工具的定义、注册与执行过程。

import json
import os
from typing import Any
from openai import OpenAI


# =============================================================================
# 1. 定义工具函数
# =============================================================================
def get_weather(city: str) -> str:
    """获取指定城市的天气信息

    Args:
        city: 城市名称，如 "beijing" 或 "shanghai"

    Returns:
        天气信息字符串
    """
    weather_data = {
        "beijing": "北京：晴天，气温 18°C，空气质量良好，东风 3 级",
        "shanghai": "上海：多云，气温 22°C，湿度 65%，南风 2 级",
    }
    city_lower = city.lower()
    if city_lower in weather_data:
        return weather_data[city_lower]
    return f"抱歉，没有 {city} 的天气数据"


# =============================================================================
# 2. 定义工具的 JSON Schema (告诉 LLM 有哪些工具可用)
# =============================================================================
tools_schema: list[Any] = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的天气信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，如 beijing 或 shanghai",
                    }
                },
                "required": ["city"],
            },
        },
    }
]


# =============================================================================
# 3. 工具执行器
# =============================================================================
def execute_tool(tool_name: str, arguments: dict) -> str:
    """执行工具函数"""
    available_tools = {
        "get_weather": get_weather,
    }

    if tool_name not in available_tools:
        return f"错误：未知的工具 {tool_name}"

    tool_func = available_tools[tool_name]
    try:
        result = tool_func(**arguments)
        return result
    except Exception as e:
        return f"工具执行错误: {str(e)}"


# =============================================================================
# 4. 主流程
# =============================================================================
def main():
    print("=" * 60)
    print(" LLM 工具调用基础示例")
    print("=" * 60)

    client = OpenAI(
        api_key=os.environ.get("OPENAI_API_KEY", "sk-xxx"),
        base_url=os.environ.get("OPENAI_BASE_URL", "https://api.openai.com/v1"),
    )

    model = os.environ.get("OPENAI_MODEL", "gpt-5.1")

    user_question = "北京和上海今天的天气怎么样？"

    print(f"\n[用户问题] {user_question}\n")

    messages: list[Any] = [
        {
            "role": "system",
            "content": "你是一个有用的助手。当用户询问天气时，使用 get_weather 工具获取信息。",
        },
        {
            "role": "user",
            "content": user_question,
        },
    ]

    print("-" * 60)
    print("Step 1: 发送请求给 LLM (带工具定义)")
    print("-" * 60)

    response = client.chat.completions.create(
        model=model,
        messages=messages,
        tools=tools_schema,
        tool_choice="auto",
    )

    assistant_message = response.choices[0].message
    print(f"\n[LLM 响应]")
    print(f"  content: {assistant_message.content}")
    print(f"  tool_calls: {assistant_message.tool_calls}")

    if assistant_message.tool_calls:
        print(f"\n[LLM 决定调用工具]")
        for tool_call in assistant_message.tool_calls:
            print(f"  - 工具名: {tool_call.function.name}")  # type: ignore
            print(f"    参数: {tool_call.function.arguments}")  # type: ignore

        print("\n" + "-" * 60)
        print("Step 2: 执行工具调用")
        print("-" * 60)

        messages.append(assistant_message)

        for tool_call in assistant_message.tool_calls:
            tool_name = tool_call.function.name  # type: ignore
            arguments = json.loads(tool_call.function.arguments)  # type: ignore

            print(f"\n[执行工具] {tool_name}({arguments})")
            tool_result = execute_tool(tool_name, arguments)
            print(f"[工具返回] {tool_result}")

            messages.append(
                {
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": tool_result,
                }
            )

        print("\n" + "-" * 60)
        print("Step 3: 将工具结果发回 LLM，生成最终回答")
        print("-" * 60)

        final_response = client.chat.completions.create(
            model=model,
            messages=messages,
        )

        final_answer = final_response.choices[0].message.content
        print(f"\n[最终回答]\n{final_answer}")

    else:
        print("\n[LLM 没有调用工具，直接回答]")
        print(assistant_message.content)

    print("\n" + "=" * 60)
    print(" 完成!")
    print("=" * 60)


if __name__ == "__main__":
    main()

如下是运行该示例的输出结果：

# python tool_calling_basic.py
============================================================
 LLM 工具调用基础示例
============================================================

[用户问题] 北京和上海今天的天气怎么样？

------------------------------------------------------------
Step 1: 发送请求给 LLM (带工具定义)
------------------------------------------------------------

[LLM 响应]
  content: None
  tool_calls: [ChatCompletionMessageFunctionToolCall(id='call_Jo93z53TOVRNLY6iKazndI9y', function=Function(arguments='{"city": "beijing"}', name='get_weather'), type='function'), ChatCompletionMessageFunctionToolCall(id='call_RO2ceN41clKgEstb2pWBHSBs', function=Function(arguments='{"city": "shanghai"}', name='get_weather'), type='function')]

[LLM 决定调用工具]
  - 工具名: get_weather
    参数: {"city": "beijing"}
  - 工具名: get_weather
    参数: {"city": "shanghai"}

------------------------------------------------------------
Step 2: 执行工具调用
------------------------------------------------------------

[执行工具] get_weather({'city': 'beijing'})
[工具返回] 北京：晴天，气温 18°C，空气质量良好，东风 3 级

[执行工具] get_weather({'city': 'shanghai'})
[工具返回] 上海：多云，气温 22°C，湿度 65%，南风 2 级

------------------------------------------------------------
Step 3: 将工具结果发回 LLM，生成最终回答
------------------------------------------------------------

[最终回答]
北京：晴天，气温 18°C，空气质量良好，东风 3 级。
上海：多云，气温 22°C，湿度 65%，南风 2 级。

============================================================
 完成!
============================================================

以上例子只是简单地将一个 Python 函数封装成可供 LLM 调用的工具，而随着 Agent 技术发展到现在，工具调用 的形态也得到了极大扩展，例如：

• 通过 MCP 协议可以调用外部工具
• 通过 Skill 系统，Agent 能够实现复杂业务逻辑的自我沉淀与复用

尽管 工具调用 在工程实现上不断扩展， 但其本质仍然是依靠 LLM 的 function calling 能力来实现的。

Toolkit 工具类

AgentScope 通过 Toolkit 类来统一管理 工具调用。如下是一个极简的例子，展示了如何将一个自定义 Python 函数添加到 Toolkit 中：

import json
from agentscope.tool import Toolkit

def basic_types(
    name: str,
    age: int,
    score: float,
    is_active: bool,
) -> None:
    pass


toolkit = Toolkit()
toolkit.register_tool_function(basic_types)

schema = toolkit.get_json_schemas()[0]

print(json.dumps(schema, indent=2, ensure_ascii=False))

生成的 json schema 如下所示：

{
  "type": "function",
  "function": {
    "name": "basic_types",
    "parameters": {
      "properties": {
        "name": {
          "type": "string"
        },
        "age": {
          "type": "integer"
        },
        "score": {
          "type": "number"
        },
        "is_active": {
          "type": "boolean"
        }
      },
      "required": [
        "name",
        "age",
        "score",
        "is_active"
      ],
      "type": "object"
    }
  }
}

可以看到，通过 Toolkit 的封装，我们可以直接将一个 Python 函数转换成 LLM 可以调用的工具，而无需手动编写繁琐的 JSON Schema，Toolkit 内部会自动根据函数签名/文档信息生成 LLM 所需的 JSON Schema。

通过 Toolkit 将普通的 Python 函数注册为 LLM 可调用的工具只是 Toolkit 功能的一部分，Toolkit 类是 AgentScope 中工具生态的 总管家，核心目标是统一化、标准化、可扩展地管理所有工具相关能力：

• 统一化：将普通函数、MCP 客户端工具、Agent Skill 等不同类型的工具纳入同一管理体系
• 标准化：为工具调用提供统一的流式响应接口、参数校验、异常处理机制
• 可扩展：支持中间件、扩展模型（extended model）、工具组（Tool Group）等功能

如下是 Toolkit 类的 __init__ 方法：

class Toolkit(StateModule):
    def __init__(
        self,
        agent_skill_instruction: str | None = None,
        agent_skill_template: str | None = None,
    ) -> None:
        super().__init__()

        # 定义所有的 tools, groups, skills, middlewares
        self.tools: dict[str, RegisteredToolFunction] = {}
        self.groups: dict[str, ToolGroup] = {}
        self.skills: dict[str, AgentSkill] = {}
        self._middlewares: list = []  # Store registered middlewares

        # 设置 AGENT SKILL 相关的提示词
        # 使得 Agent 背后的 LLM 能够理解 SKILL 的概念，并动态调用某个 SKILL
        self._agent_skill_instruction = (
            agent_skill_instruction or self._DEFAULT_AGENT_SKILL_INSTRUCTION
        )
        # 每个具体的 skill 定义所遵守的模版
        self._agent_skill_template = (
            agent_skill_template or self._DEFAULT_AGENT_SKILL_TEMPLATE
        )

接下来将详细介绍 Toolkit 的核心实现原理。

工具注册机制

工具注册是 Toolkit 最基础的能力，核心函数是 register_tool_function() 方法：

def register_tool_function(
        self,
        tool_func: ToolFunction,
        group_name: str | Literal["basic"] = "basic",
        preset_kwargs: dict[str, JSONSerializableObject] | None = None,
        func_name: str | None = None,
        func_description: str | None = None,
        json_schema: dict | None = None,
        include_long_description: bool = True,
        include_var_positional: bool = False,
        include_var_keyword: bool = False,
        postprocess_func: (
            Callable[
                [ToolUseBlock, ToolResponse],
                ToolResponse | None,
            ]
            | Callable[
                [ToolUseBlock, ToolResponse],
                Awaitable[ToolResponse | None],
            ]
        )
        | None = None,
        namesake_strategy: Literal[
            "override",
            "skip",
            "raise",
            "rename",
        ] = "raise",
    ) -> None:

这个方法的唯一必选参数是 tool_func，即要注册的 Python 函数，这个 Python 函数可以是 同步函数、异步函数、生成器函数、异步生成器函数、协程 等多种类型。而其他可选参数都是用于定义注册时的行为的，具体来说：

• group_name：工具所属的组名，默认是 basic，basic 组中的工具总是会被注册到 LLM 调用的 tools 列表中，而其他组的工具只有在激活时才会被注册
• preset_kwargs：预设的关键字参数，这些参数不会包含在 JSON schema 中，因此 LLM 看不到这些参数。相反这些参数会在调用函数时自动设置
• func_name：手动提供的函数名称，没有提供时将自动从函数签名/文档中提取
• func_description：手动提供的函数描述，没有提供时将自动从函数签名/文档中提取
• json_schema：手动提供的 JSON schema，如果没有提供将自动从函数签名/文档中提取
• include_var_positional：是否包含可变的位置参数，即通常的 *args，默认 False
• include_var_keyword：是否包含可变的关键词参数，即通常的 **kwargs，默认 False
• postprocess_func：工具调用结束后的 hook 函数
• namesake_strategy：同名工具函数的处理策略，默认是 raise 即抛出异常，其他值包括：override（覆盖）、skip（忽略）和 rename（重命名，即在函数名后添加一个随机字符串作为后缀）

register_tool_function 会处理不同类型的函数，包括普通函数、partial（partial 函数绑定的参数等同于 preset_kwargs）、MCPToolFunction（MCP 工具函数，也是可调用对象），核心目标是获取这些函数的 JSON schema，并将注册的函数用 RegisteredToolFunction 对象表示，保存到 self.tools 这个 map 中

func_obj = RegisteredToolFunction(
    name=func_name,
    group=group_name,
    source="function",
    original_func=original_func,
    json_schema=json_schema, # 原始函数
    preset_kwargs=preset_kwargs or {},
    original_name=original_name,
    extended_model=None,
    mcp_name=mcp_name,
    postprocess_func=postprocess_func,  # 后置 hook 处理函数
)

self.tools[func_name] = func_obj

_parse_tool_function() 负责提取工具函数的 JSON schema，它的核心原理是根据函数参数的类型注解、docstring 等信息生成工具函数的 JSON schema，包含函数功能的描述、参数描述等信息。

ToolFunction 类型说明

框架代码使用 ToolFunction 类型来表示可被注册的工具函数的类型，因为这里涉及一些 Python 基础知识，花点时间解释一下：

ToolFunction = Callable[
    ...,
    Union[
        # sync function
        ToolResponse,
        # async function
        Awaitable[ToolResponse],
        # sync generator function
        Generator[ToolResponse, None, None],
        # async generator function
        AsyncGenerator[ToolResponse, None],
        # async function that returns async generator
        Coroutine[Any, Any, AsyncGenerator[ToolResponse, None]],
        # async function that returns sync generator
        Coroutine[Any, Any, Generator[ToolResponse, None, None]],
    ],
]

ToolFunction 定义表示 参数可以是任意类型，返回值可以是多种类型的可调用对象。具体来说，返回值可以是以下情况：

• ToolResponse：普通的同步函数，直接返回 ToolResponse

def sync_tool() -> ToolResponse:
    """普通同步函数"""
    # 模拟快速处理
    result = "计算完成：1 + 1 = 2"
    return ToolResponse(result)

• Awaitable[ToolResponse]：使用 async def 定义的异步函数，调用它会返回一个协程（Coroutine），可以用 await 来获取最终的 ToolResponse。因此这种函数类型里，返回值类型应该写成 Awaitable[ToolResponse]（Coroutine 是一种具体的 Awaitable 实现）。这里有一点需要注意，由于 Python 类型提示的语法糖，在实际定义时，可以简单写为该协程运行的结果类型，即这里的 ToolResponse

# 这在逻辑上等同于如下定义，但比较啰嗦
# def async_tool_raw() -> Coroutine[Any, Any, ToolResponse]:
async def async_tool() -> ToolResponse:
    """异步函数"""
    # 模拟网络请求或数据库查询 (非阻塞)
    await asyncio.sleep(1)
    result = "API数据拉取成功"
    return ToolResponse(result)

• Generator[ToolResponse, None, None] ：同步生成器函数，使用 def 定义，内部包含 yield 关键字。它能多次产出（yield）结果，形成数据流

def sync_generator_tool() -> Generator[ToolResponse, None, None]:
    """同步生成器函数"""
    chunks = ["第一段内容", "第二段内容", "第三段内容"]
    for chunk in chunks:
        # 逐步产出数据
        yield ToolResponse(chunk)

# 调用方式：
# for response in sync_generator_tool():
#     print(response)

• AsyncGenerator[ToolResponse, None]：使用 async def 定义，且内部包含 yield。允许你在生成数据的流式过程中使用 await。这里需要注意，虽然说使用 async def 定义的函数会返回一个协程，但是如果函数实现中包含 yield 关键字，这个函数就不再是普通的异步函数，而是一个 异步生成器函数，此时返回的是一个 异步生成器 (AsyncGenerator)，因此要用 AsyncGenerator[ToolResponse, None] 来标注其返回类型。

async def async_generator_tool() -> AsyncGenerator[ToolResponse, None]:
    """异步生成器函数"""
    chunks = ["第一段流式数据", "第二段流式数据", "第三段流式数据"]
    for chunk in chunks:
        await asyncio.sleep(0.5) # 模拟异步延迟（如等待网络包）
        yield ToolResponse(chunk)

# 调用方式：
# async for response in async_generator_tool():
#     print(response)

• Coroutine[Any, Any, AsyncGenerator[ToolResponse, None]]：返回异步生成器的异步函数，首先是一个异步函数，该异步函数返回一个异步生成器。同样由于类型提示的语法糖，函数定义时结果类型标注为 AsyncGenerator[ToolResponse, None] 即可

async def async_func_returning_async_gen() -> AsyncGenerator[ToolResponse, None]:
    """异步函数返回异步生成器"""

    # 1. 前置的异步操作（例如：异步登录、握手获取 Token）
    print("正在建立异步连接...")
    await asyncio.sleep(1)

    # 2. 定义内部的异步生成器
    async def inner_stream() -> AsyncGenerator[ToolResponse, None]:
        for i in range(3):
            await asyncio.sleep(0.1)
            yield ToolResponse(f"流式数据包 {i}")

    # 3. 返回这个生成器对象 (注意这里是 return，不是 yield)
    return inner_stream()

# 调用方式：
# gen = await async_func_returning_async_gen() # 先 await 拿到生成器
# async for response in gen:                   # 再 async for 读取流
#     print(response)

• Coroutine[Any, Any, Generator[ToolResponse, None, None]]：返回同步生成器的异步函数。同样由于类型提示的语法糖，函数定义时结果类型标注为 Generator[ToolResponse, None] 即可

async def async_func_returning_sync_gen() -> Generator[ToolResponse, None, None]:
    """异步函数返回同步生成器"""

    # 1. 异步前置操作 (例如从 S3 下载了一个文件到本地)
    print("正在异步下载文件...")
    await asyncio.sleep(1)

    # 2. 定义内部的同步生成器 (读取本地文件不需要 async)
    def inner_stream() -> Generator[ToolResponse, None, None]:
        local_data = ["本地数据块A", "本地数据块B"]
        for data in local_data:
            yield ToolResponse(data)

    # 3. 返回同步生成器
    return inner_stream()

# 调用方式：
# gen = await async_func_returning_sync_gen() # 先 await 拿到同步生成器
# for response in gen:                        # 注意这里是普通的 for 循环
#     print(response)

这些函数类型覆盖了同步/异步、流式/非流式的所有场景，使得各种函数都可以被注册为 AgentScope 框架的工具。

工具组

Toolkit 还引入了工具组的概念，可以将相关的工具进行批量管理，而且支持 Agent 动态调整激活的工具组，这种特性使得 Agent 可以根据需求动态开启/关闭某些工具，减少工具描述对 Context 的占用。

• Toolkit 默认提供一个 basic 工具组，该工具组始终处于激状状态，而且无法删除
• 注册工具时，如果没有指定 group，则默认在 “basic” 工具组中

create_tool_group() 方法负责创建工具组，它其实就是在 self.groups 中添加一个 ToolGroup 对象：

self.groups[group_name] = ToolGroup(
    name=group_name,
    description=description,
    notes=notes,
    active=active,
)

update_tool_groups() 可以更新工具组的 active 状态，remove_tool_groups 则用于删除工具组，此时 self.tools 中所有属于该 group 的工具都会被移除。

那到底如何实现工具组的动态管理功能呢，创建 ReActAgent 时提供了 enable_meta_tool 配置参数，如果设置为 True，则该 Agent 会注册一个 reset_equipped_tools 工具，通过该工具允许 Agent 动态调整装备的工具组。

if enable_meta_tool:
    self.toolkit.register_tool_function(
        self.toolkit.reset_equipped_tools,
    )

Toolkit.reset_equipped_tools 方法的实现如下：

# 每次都全量重新设置要开启的 Group
def reset_equipped_tools(self, **kwargs: Any) -> ToolResponse:
    """This function allows you to activate or deactivate tool groups
    dynamically based on your current task requirements.
    ......"""

    # 先将所有 tool_group 都设置为 active False
    self.update_tool_groups(list(self.groups.keys()), active=False)

    # 激活
    to_activate = []
    for key, value in kwargs.items():
        if value:
            to_activate.append(key)
    self.update_tool_groups(to_activate, active=True)

    # 添加各个 active group 的 note
    notes = self.get_activated_notes()

    # 构造工具响应
    text_response = ""
    if to_activate:
        text_response += (
            "Now tool groups "
            + ", ".join([f"'{_}'" for _ in to_activate])
            + " are activated."
        )

    # 设置 note
    if notes:
        text_response += (
            f" You MUST follow these notes to use these tools:\n"
            f"<notes>{notes}</notes>"
        )

    if not text_response:
        text_response = "All tool groups are now deactivated currently."

    # 返回响应
    return ToolResponse(
        content=[
            TextBlock(
                type="text",
                text=text_response,
            ),
        ],
    )

这就是 AgentScope 支持动态调整工具组的实现原理。Toolkit 的工具组设计，让 Agent 从 一次性携带所有工具 的笨重模式，变成了 按需装备、动态切换 的轻量化模式。既降低了上下文负担，又提升了工具调用准确率。

extended_model

不知道大家有没有发现，动态重置工具组的实现有一个问题：reset_equipped_tools 的参数是一个 **kwargs，这意味着注册 reset_equipped_tools 工具时，其实是无法函数参数的形式来告诉 LLM 当前有哪些 Group，LLM 也就无法构造出准确的调用参数。那 AgentScope 如何解决这个问题呢？

每个 RegisteredToolFunction 都会有一个 extended_model 字段：

@dataclass
class RegisteredToolFunction:
    extended_model: Type[BaseModel] | None = None

当获取工具的 extended_json_schema 特性时，就会将 RegisteredToolFunction 本身解析到的 json schema 与 extended_model 进行进行合并，最后得到该工具的完整 json schema。

@property
def extended_json_schema(self) -> dict:
    """Get the JSON schema of the tool function, if an extended model is
    set, the merged JSON schema will be returned."""

而 reset_equipped_tools 工具就是依靠 extended_model 来动态提供当前的 group 信息的。当调用 Toolkit.get_json_schemas() 时，其会根据当前 active 的 Group，动态生成一个 extended_model，并设置到 reset_equipped_tools 工具中：

def get_json_schemas(
    self,
) -> list[dict]:
    if "reset_equipped_tools" in self.tools:
        fields = {}
        for group_name, group in self.groups.items():
            if group_name == "basic":
                continue
            fields[group_name] = (
                bool,
                Field(
                    default=False,
                    description=group.description,
                ),
            )
        # 对 reset_equipped_tools 工具函数设置 extended model
        # extend model 为包含所有 group 的描述信息
        extended_model = create_model("_DynamicModel", **fields)
        self.set_extended_model(
            "reset_equipped_tools",
            extended_model,
        )

这样最终 reset_equipped_tools 工具的 extended_json_schema 就包含了所有 Group 的信息，LLM 在调用时就可以正确设置各个 group 的 active 状态了。

extended_model 是 Toolkit 中用于动态扩展/覆盖工具函数 JSON Schema 的核心机制，本质是通过 自定义 Pydantic BaseModel 来补充/替换工具函数自动生成的 Schema 规则，解决 自动生成的 Schema 无法满足复杂场景需求 的问题。

工具调用

接下来我们再来看下 Toolkit 是怎么调用工具的。当 LLM 生成工具调用指令时，AgentScope 框架会生成一个 ToolUseBlock 对象，用来表示 工具调用请求，ReActAgent 在其 _acting() 方法中调用 Toolkit.call_tool_function 执行该工具调用。

@trace_toolkit
@_apply_middlewares
async def call_tool_function(
    self,
    tool_call: ToolUseBlock,
) -> AsyncGenerator[ToolResponse, None]:
    """Execute the tool function by the `ToolUseBlock` and return the
    tool response chunk in unified streaming mode, i.e. an async
    generator of `ToolResponse` objects.
    ......

call_tool_function 的核心逻辑就是根据工具名称，找到对应的 RegisteredToolFunction 对象，之后填充参数（LLM 提供的参数 + 预设参数），并调用真正的 Python 函数：

tool_func = self.tools[tool_call["name"]]

# LLM 提供的参数 + 预设参数
kwargs = {
    **tool_func.preset_kwargs,
    **(tool_call.get("input", {}) or {}),
}

res = tool_func.original_func(**kwargs)

注意到，call_tool_function 函数的返回值总是一个 AsyncGenerator[ToolResponse, None]，即总是以异步流式的方式返回工具调用结果。而我们之前说过，真正的工具函数可以是各种形式（同步/异步、流式/非流式），所以需要对工具函数的返回值进行包装，统一转化为 AsyncGenerator[ToolResponse, None]，这样在调用处就可以统一异步流式来处理工具调用结果：

tool_res = await self.toolkit.call_tool_function(tool_call)

# Async generator handling
async for chunk in tool_res:
    ......

为了实现这一转换，实现了如下几个 wrapper 函数，返回值类型都是 AsyncGenerator[ToolResponse, None]

• _object_wrapper()
• _sync_generator_wrapper()
• _async_generator_wrapper()

工具调用还有一点需要注意，每个 RegisteredToolFunction 都会有可选的 postprocess_func，用来对工具调用的结果进行后处理，因此 call_tool_function() 中还有这部分相关逻辑，用来对工具函数返回的每一个 ToolResponse 进行自定义操作。如下以 _sync_generator_wrapper 的实现为例：

async def _sync_generator_wrapper(
    sync_generator: Generator[ToolResponse, None, None],
    postprocess_func: (
        Callable[[ToolResponse], ToolResponse | None]
        | Callable[[ToolResponse], Awaitable[ToolResponse | None]]
    )
    | None,
) -> AsyncGenerator[ToolResponse, None]:
    """Wrap a sync generator to an async generator."""
    for chunk in sync_generator:
        yield await _postprocess_tool_response(chunk, postprocess_func)

工具调用中间件

AgentScope 的工具调用还实现了中间件（middleware）机制，允许开发者对工具调用的执行过程进行拦截和修改。中间件系统遵循 洋葱模型，每个中间件包裹在前一个中间件之外，形成层次结构。这使得开发者可以可以在工具调用前后都插入自定义处理逻辑。

当注册顺序为 M1 → M2 → ... → Mn 时，执行顺序为：

• 预处理（调用工具函数前的处理）: M1 → M2 → … → Mn → 实际函数
• 后处理（得到工具函数结果后的处理）: 实际函数 → Mn → … → M2 → M1

例如对于如下代码：

toolkit.register_middleware(middleware_1)  # 先注册
toolkit.register_middleware(middleware_2)  # 后注册

执行顺序为：

┌─────────────────────────────────────────────────────────────────┐
│                                                                 │
│  middleware_1 预处理                                             │
│      ↓                                                          │
│  middleware_2 预处理                                             │
│      ↓                                                          │
│  call_tool_function 核心逻辑                                      │
│      ↓                                                          │
│  middleware_2 后处理                                             │
│      ↓                                                          │
│  middleware_1 后处理                                             │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

工具调用的中间件机制就是靠 @_apply_middlewares 这个装饰器来实现的：

@trace_toolkit
@_apply_middlewares
async def call_tool_function(
    self,
    tool_call: ToolUseBlock,
) -> AsyncGenerator[ToolResponse, None]:

def _apply_middlewares(func):
    @wraps(func)
    async def wrapper(self: "Toolkit", tool_call: ToolUseBlock):
        # 1. 获取中间件列表
        middlewares = getattr(self, "_middlewares", [])

        # 2. 无中间件时直接执行原函数
        if not middlewares:
            async for chunk in await func(self, tool_call):
                yield chunk
            return

        # 3. 构建基础 handler (最内层)
        async def base_handler(**kwargs):
            return await func(self, **kwargs)

        # 4. 反向构建中间件链
        current_handler = base_handler
        for middleware in reversed(middlewares):
            def make_handler(mw, handler):
                async def wrapped(**kwargs):
                    return mw(kwargs, handler)  # 中间件签名: (kwargs, next_handler)
                return wrapped
            current_handler = make_handler(middleware, current_handler)

        # 5. 执行中间件链
        async for chunk in await current_handler(tool_call=tool_call):
            yield chunk

    return wrapper

而 Toolkit 提供了 register_middleware 方法来注册中间件：

class Toolkit:
    def __init__(self):
        self._middlewares: list = []  # 存储中间件

    def register_middleware(self, middleware: Callable):
        """注册洋葱式中间件"""
        self._middlewares.append(middleware)

中间件应该有如下签名：

async def middleware(
    kwargs: dict,
    next_handler: Callable,
) -> AsyncGenerator[ToolResponse, None]:
    # 从 kwargs 访问参数
    tool_call = kwargs["tool_call"]

    # 预处理
    # ...

    # 调用下一个中间件或工具函数
    async for response in await next_handler(**kwargs):
        # 后处理
        yield response

• kwargs：上下文参数。当前仅包含 tool_call (ToolUseBlock)
• next_handler：下一个中间件或工具函数的引用
• 返回值：AsyncGenerator[ToolResponse, None]，和工具函数返回值类型一致

小结

这篇文章我们重点分析了 AgentScope 框架实现 Agent 工具调用的核心类 Toolkit，包括其工具注册、动态工具组、工具调用、中间件等关键功能。下一篇文章我们将继续分析 AgentScope 如何实现 MCP、Agent Skill 等功能，这些功能也可以认为是 Agent 工具能力的一部分。

Reference

• AgentScope 工具
• AgentScope 中间件