注意
Go to the end 下载完整示例代码。
工具¶
为确保精确可靠的工具解析，AgentScope 全面采用工具API，并提供以下特性：
支持从Python函数及其文档字符串中自动解析工具
同时支持同步与异步工具函数
支持流式工具响应（同步或异步生成器）
支持对工具 JSON 模式的动态扩展
支持使用适当的信号处理中断工具执行
支持智能体的自主工具管理
所有上述功能都是由 AgentScope 中的 Toolkit 类实现的，该类负责管理工具函数及其执行。
提示
对于 MCP（Model Context Protocol）的支持，请参阅 MCP 部分。
import asyncio
import inspect
import json
from typing import Any, AsyncGenerator
from pydantic import BaseModel, Field
import agentscope
from agentscope.message import TextBlock, ToolUseBlock
from agentscope.tool import ToolResponse, Toolkit, execute_python_code
工具函数¶
在AgentScope中，工具功能是一个Python函数，它
返回一个
ToolResponse对象或一个生成ToolResponse对象的生成器包含一个描述工具功能和参数的文档字符串
一个工具函数的模板如下：
def tool_function(a: int, b: str) -> ToolResponse:
"""{function description}
Args:
a (int):
{description of the first parameter}
b (str):
{description of the second parameter}
"""
提示
实例方法和类方法也可以用作工具函数，而 self 和 cls 参数将被忽略。
AgentScope 在 agentscope.tool 模块下提供了多个内置工具函数，如 execute_python_code、execute_shell_command 以及文本文件读写功能。
print("Built-in Tool Functions:")
for _ in agentscope.tool.__all__:
if _ not in ["Toolkit", "ToolResponse"]:
print(_)
Built-in Tool Functions:
execute_python_code
execute_shell_command
view_text_file
write_text_file
insert_text_file
dashscope_text_to_image
dashscope_text_to_audio
dashscope_image_to_text
openai_text_to_image
openai_text_to_audio
openai_edit_image
openai_create_image_variation
openai_image_to_text
openai_audio_to_text
工具集¶
Toolkit 类旨在管理工具函数，从文档字符串中提取它们的JSON模式，并为工具执行提供统一的接口。
基本用法¶
Toolkit 类的基本功能是注册工具函数并执行它们。
# Prepare a custom tool function
async def my_search(query: str, api_key: str) -> ToolResponse:
"""A simple example tool function.
Args:
query (str):
The search query.
api_key (str):
The API key for authentication.
"""
return ToolResponse(
content=[
TextBlock(
type="text",
text=f"Searching for '{query}' with API key '{api_key}'",
),
],
)
# Register the tool function in a toolkit
toolkit = Toolkit()
toolkit.register_tool_function(my_search)
注册工具函数时，可以通过调用get_json_schemas方法获取其JSON模式。
print("Tool JSON Schemas:")
print(json.dumps(toolkit.get_json_schemas(), indent=4, ensure_ascii=False))
Tool JSON Schemas:
[
{
"type": "function",
"function": {
"name": "my_search",
"parameters": {
"properties": {
"query": {
"description": "The search query.",
"type": "string"
},
"api_key": {
"description": "The API key for authentication.",
"type": "string"
}
},
"required": [
"query",
"api_key"
],
"type": "object"
},
"description": "A simple example tool function."
}
}
]
Toolkit 也允许开发者预设工具函数的参数，尤其适用于API密钥或其他敏感信息。
# Clear the toolkit first
toolkit.clear()
# Register tool function with preset keyword arguments
toolkit.register_tool_function(my_search, preset_kwargs={"api_key": "xxx"})
print("Tool JSON Schemas with Preset Arguments:")
print(json.dumps(toolkit.get_json_schemas(), indent=4, ensure_ascii=False))
Tool JSON Schemas with Preset Arguments:
[
{
"type": "function",
"function": {
"name": "my_search",
"parameters": {
"properties": {
"query": {
"description": "The search query.",
"type": "string"
}
},
"required": [
"query"
],
"type": "object"
},
"description": "A simple example tool function."
}
}
]
在 Toolkit 中，call_tool_function 方法接受一个工具使用块作为输入并执行对应的工具函数，返回 一个统一的异步生成器，该生成器会生成 ToolResponse 对象。
async def example_tool_execution() -> None:
"""Example of executing a tool call."""
res = await toolkit.call_tool_function(
ToolUseBlock(
type="tool_use",
id="123",
name="my_search",
input={"query": "AgentScope"},
),
)
# Only one tool response is expected in this case
print("Tool Response:")
async for tool_response in res:
print(tool_response)
asyncio.run(example_tool_execution())
Tool Response:
ToolResponse(content=[{'type': 'text', 'text': "Searching for 'AgentScope' with API key 'xxx'"}], metadata=None, stream=False, is_last=True, is_interrupted=False, id='2025-09-08 07:57:58.121_386e48')
动态扩展JSON模式¶
Toolkit 允许通过调用 set_extended_model 方法动态扩展工具函数的 JSON 模式。
此特性允许在不修改原有定义的情况下为工具函数添加更多参数。
提示
相关场景包括动态结构化输出和CoT（思维链）推理
注意
要扩展的函数应接受可变关键字参数（**kwargs），以便可以将额外字段传递给它。
以CoT推理为例，我们可以用thinking字段扩展所有工具函数，让智能体总结当前状态，然后决定下一步操作。
# Example tool function
def tool_function(**kwargs: Any) -> ToolResponse:
"""A tool function"""
return ToolResponse(
content=[
TextBlock(
type="text",
text=f"Received parameters: {kwargs}",
),
],
)
# Add a thinking field so that the agent could think before giving the other parameters.
class ThinkingModel(BaseModel):
"""A Pydantic model for additional fields."""
thinking: str = Field(
description="Summarize the current state and decide what to do next.",
)
# Register
toolkit.set_extended_model("my_search", ThinkingModel)
print("The extended JSON Schema:")
print(json.dumps(toolkit.get_json_schemas(), indent=4, ensure_ascii=False))
The extended JSON Schema:
[
{
"type": "function",
"function": {
"name": "my_search",
"parameters": {
"properties": {
"query": {
"description": "The search query.",
"type": "string"
},
"thinking": {
"description": "Summarize the current state and decide what to do next.",
"type": "string"
}
},
"required": [
"query",
"thinking"
],
"type": "object"
},
"description": "A simple example tool function."
}
}
]
中断工具执行¶
Toolkit 类支持 异步工具函数 的 执行中断，并提供了一个全面的 智能体导向后处理机制。
此类中断基于 asyncio 取消机制实现，后处理则根据工具函数的返回类型而有所不同。
注意
对于同步工具函数，其执行不能被asyncio取消操作中断。因此中断处理是在智能体内部而非工具包中进行的。 更多信息请参阅Agent章节。
Specifically, if the tool function returns a ToolResponse object, a predefined ToolResponse object with an interrupted message will be yielded.
So that the agent can observe the interruption and handle it accordingly.
Besides, a flag is_interrupted will be set to True in the response, and the external caller can decide whether to throw the CancelledError exception to the outer layer.
一个可被中断的异步工具函数示例如下：
async def non_streaming_function() -> ToolResponse:
"""A non-streaming tool function that can be interrupted."""
await asyncio.sleep(1) # Simulate a long-running task
# Fake interruption for demonstration
raise asyncio.CancelledError()
# The following code won't be executed due to the cancellation
return ToolResponse(
content=[
TextBlock(
type="text",
text="Run successfully!",
),
],
)
async def example_tool_interruption() -> None:
"""Example of tool interruption."""
toolkit = Toolkit()
toolkit.register_tool_function(non_streaming_function)
res = await toolkit.call_tool_function(
ToolUseBlock(
type="tool_use",
id="123",
name="non_streaming_function",
input={},
),
)
async for tool_response in res:
print("Tool Response:")
print(tool_response)
print("The interrupted flag:")
print(tool_response.is_interrupted)
asyncio.run(example_tool_interruption())
Tool Response:
ToolResponse(content=[{'type': 'text', 'text': '<system-info>The tool call has been interrupted by the user.</system-info>'}], metadata=None, stream=True, is_last=True, is_interrupted=True, id='2025-09-08 07:57:59.125_f58776')
The interrupted flag:
True
对于流式工具函数，它返回一个异步生成器，Toolkit 会将中断消息附加到响应的前一个数据块上。
通过这种方式，智能体可以观察到工具在中断前返回了什么。
中断一个流式工具函数的示例如下：
async def streaming_function() -> AsyncGenerator[ToolResponse, None]:
"""A streaming tool function that can be interrupted."""
# Simulate a chunk of response
yield ToolResponse(
content=[
TextBlock(
type="text",
text="1234",
),
],
)
# Simulate interruption
raise asyncio.CancelledError()
# The following code won't be executed due to the cancellation
yield ToolResponse(
content=[
TextBlock(
type="text",
text="123456789",
),
],
)
async def example_streaming_tool_interruption() -> None:
"""Example of streaming tool interruption."""
toolkit = Toolkit()
toolkit.register_tool_function(streaming_function)
res = await toolkit.call_tool_function(
ToolUseBlock(
type="tool_use",
id="xxx",
name="streaming_function",
input={},
),
)
i = 0
async for tool_response in res:
print(f"Chunk {i}:")
print(tool_response)
print("The interrupted flag: ", tool_response.is_interrupted, "\n")
i += 1
asyncio.run(example_streaming_tool_interruption())
Chunk 0:
ToolResponse(content=[{'type': 'text', 'text': '1234'}], metadata=None, stream=False, is_last=True, is_interrupted=False, id='2025-09-08 07:57:59.128_2a0723')
The interrupted flag: False
Chunk 1:
ToolResponse(content=[{'type': 'text', 'text': '1234'}, {'type': 'text', 'text': '<system-info>The tool call has been interrupted by the user.</system-info>'}], metadata=None, stream=False, is_last=True, is_interrupted=True, id='2025-09-08 07:57:59.128_2a0723')
The interrupted flag: True
自动化工具管理¶
Toolkit 类通过引入工具组的概念以及一个名为reset_equipped_tools的元工具函数来支持自动工具管理。
工具组是一组相关的工具功能，例如浏览器使用工具、地图服务工具等，这些功能将一起管理。
只有在已激活的组中的工具才会对智能体可见，即可通过toolkit.get_json_schemas()方法访问。
注意有一个特殊的组叫做 basic，它总是被激活，且未指定组名注册的工具将默认被添加到该组。
提示
basic 组合确保在您不需要组合功能的情况下，工具的基本使用不会受到影响。
现在我们尝试创建一个名为 browser_use 的工具组，其中包含一些网页浏览工具。
def navigate(url: str) -> ToolResponse:
"""Navigate to a web page.
Args:
url (str):
The URL of the web page to navigate to.
"""
pass
def click_element(element_id: str) -> ToolResponse:
"""Click an element on the web page.
Args:
element_id (str):
The ID of the element to click.
"""
pass
toolkit = Toolkit()
# Create a tool group named browser_use
toolkit.create_tool_group(
group_name="browser_use",
description="The tool functions for web browsing.",
active=False,
# The notes when using these tools
notes="""1. Use ``navigate`` to open a web page.
2. When requiring user authentication, ask the user for the credentials
3. ...""",
)
toolkit.register_tool_function(navigate, group_name="browser_use")
toolkit.register_tool_function(click_element, group_name="browser_use")
# We can also register some basic tools
toolkit.register_tool_function(execute_python_code)
如果我们检查工具JSON模式，我们可以只看到execute_python_code工具，因为browser_use组尚未激活:
print("Tool JSON Schemas with Group:")
print(json.dumps(toolkit.get_json_schemas(), indent=4, ensure_ascii=False))
Tool JSON Schemas with Group:
[
{
"type": "function",
"function": {
"name": "execute_python_code",
"parameters": {
"properties": {
"code": {
"description": "The Python code to be executed.",
"type": "string"
},
"timeout": {
"default": 300,
"description": "The maximum time (in seconds) allowed for the code to run.",
"type": "number"
}
},
"required": [
"code"
],
"type": "object"
},
"description": "Execute the given python code in a temp file and capture the return\n\ncode, standard output and error. Note you must `print` the output to get\nthe result, and the tmp file will be removed right after the execution."
}
}
]
使用 update_tool_groups 方法来激活或停用工具组：
toolkit.update_tool_groups(group_names=["browser_use"], active=True)
print("Tool JSON Schemas with Group:")
print(json.dumps(toolkit.get_json_schemas(), indent=4, ensure_ascii=False))
Tool JSON Schemas with Group:
[
{
"type": "function",
"function": {
"name": "navigate",
"parameters": {
"properties": {
"url": {
"description": "The URL of the web page to navigate to.",
"type": "string"
}
},
"required": [
"url"
],
"type": "object"
},
"description": "Navigate to a web page."
}
},
{
"type": "function",
"function": {
"name": "click_element",
"parameters": {
"properties": {
"element_id": {
"description": "The ID of the element to click.",
"type": "string"
}
},
"required": [
"element_id"
],
"type": "object"
},
"description": "Click an element on the web page."
}
},
{
"type": "function",
"function": {
"name": "execute_python_code",
"parameters": {
"properties": {
"code": {
"description": "The Python code to be executed.",
"type": "string"
},
"timeout": {
"default": 300,
"description": "The maximum time (in seconds) allowed for the code to run.",
"type": "number"
}
},
"required": [
"code"
],
"type": "object"
},
"description": "Execute the given python code in a temp file and capture the return\n\ncode, standard output and error. Note you must `print` the output to get\nthe result, and the tmp file will be removed right after the execution."
}
}
]
除此之外, Toolkit 提供了一个名为 reset_equipped_tools 的元工具函数, 接受当前组名作为参数来指定需要激活哪些组:
注意
在 ReActAgent 类中，你可以通过在构造函数中设置 enable_meta_tool=True 来启用元工具功能。
# Register the meta tool function
toolkit.register_tool_function(toolkit.reset_equipped_tools)
reset_equipped = next(
tool
for tool in toolkit.get_json_schemas()
if tool["function"]["name"] == "reset_equipped_tools"
)
print("JSON schema of the ``reset_equipped_tools`` function:")
print(
json.dumps(
reset_equipped,
indent=4,
ensure_ascii=False,
),
)
JSON schema of the ``reset_equipped_tools`` function:
{
"type": "function",
"function": {
"name": "reset_equipped_tools",
"parameters": {
"properties": {
"browser_use": {
"default": false,
"description": "The tool functions for web browsing.",
"type": "boolean"
}
},
"type": "object"
},
"description": "Choose appropriate tools to equip yourself with, so that you can\n\nfinish your task. Each argument in this function represents a group\nof related tools, and the value indicates whether to activate the\ngroup or not. Besides, the tool response of this function will\ncontain the precaution notes for using them, which you\n**MUST pay attention to and follow**. You can also reuse this function\nto check the notes of the tool groups.\n\nNote this function will `reset` the tools, so that the original tools\nwill be removed first."
}
}
当agent调用reset_equipped_tools功能时，对应的工具组将被激活，工具响应将包含已激活工具组的注释。
async def mock_agent_reset_tools() -> None:
"""Mock agent to reset tool groups."""
# Call the meta tool function
res = await toolkit.call_tool_function(
ToolUseBlock(
type="tool_use",
id="154",
name="reset_equipped_tools",
input={
"browser_user": True,
},
),
)
async for tool_response in res:
print("Text content in tool Response:")
print(tool_response)
asyncio.run(mock_agent_reset_tools())
Text content in tool Response:
ToolResponse(content=[{'type': 'text', 'text': "Active tool groups successfully: ['browser_user']. You MUST follow these notes to use the tools:\n<notes>## About browser_use Tools\n1. Use ``navigate`` to open a web page.\n2. When requiring user authentication, ask the user for the credentials\n3. ...</notes>"}], metadata=None, stream=False, is_last=True, is_interrupted=False, id='2025-09-08 07:57:59.136_88bea2')
该工具包还提供了收集已激活工具组注释的方法，您可以将其组装到您的智能体系统提示中。
提示
自动工具管理功能已在ReActAgent类中实现，详情请参阅Agent章节。
# Create one more tool group
toolkit.create_tool_group(
group_name="map_service",
description="The google map service tools.",
active=True,
notes="""1. Use ``get_location`` to get the location of a place.
2. ...""",
)
print("The gathered notes of the activated tool groups:")
print(toolkit.get_activated_notes())
The gathered notes of the activated tool groups:
## About browser_use Tools
1. Use ``navigate`` to open a web page.
2. When requiring user authentication, ask the user for the credentials
3. ...
## About map_service Tools
1. Use ``get_location`` to get the location of a place.
2. ...
延伸阅读¶
脚本总运行时长：（0分钟1.022秒）