跳转至

工具与审批

通过 runtime.tools 注册和执行工具。审批处理器在参数校验后运行,决定当前调用是否执行。

工具调用指南 · 交互式审批示例 · MCP 配置

常用接口

接口 用途
ToolManager 注册、选择和执行工具,管理内置工具与 MCP 连接。
ToolApprovalRequest 处理器读取 call_id、name、校验后的 arguments,以及本次调用的 workspace(绝对路径 Path)。
ToolApprovalDecision approve() 批准;reject(reason) 拒绝并把原因交给 Agent。

注册与执行

注册、选择和执行工具,管理内置工具与 MCP 连接。execute_many() 返回按输入顺序排列的结果。

ToolManager

Facade that coordinates tool registration, discovery, MCP, and execution.

tool property

tool: Callable[..., Any]

Return the registry decorator for local Python function tools.

register

register(tool: Tool) -> Tool

Register an already constructed Tool object.

register_builtin_tools

register_builtin_tools() -> list[str]

Register built-in local tools using manager-owned config and state.

connect_mcp async

connect_mcp() -> list[str]

Connect configured MCP servers and register their enabled tools.

set_mcp_server

set_mcp_server(server: MCPServerConfig) -> None

Add or replace one MCP server configuration without connecting it.

remove_mcp_server async

remove_mcp_server(server_name: str) -> None

Close and forget one MCP server configuration.

get_mcp_server_config

get_mcp_server_config(server_name: str) -> MCPServerConfig | None

Return one configured MCP server by name.

list_mcp_server_configs

list_mcp_server_configs() -> list[MCPServerConfig]

Return all configured MCP servers.

get_mcp_server_status

get_mcp_server_status(server_name: str) -> MCPServerStatus | None

Return one MCP server status by name.

list_mcp_server_statuses

list_mcp_server_statuses() -> list[MCPServerStatus]

Return runtime status for all configured MCP servers.

connect_mcp_server async

connect_mcp_server(server: MCPServerConfig | str) -> list[str]

Connect one MCP server and register its enabled tools.

reload_mcp async

reload_mcp() -> list[str]

Reconnect configured MCP servers and rebuild their registered tools.

reload_mcp_server async

reload_mcp_server(server_name: str) -> list[str]

Reconnect one MCP server and rebuild its registered tools.

sync_mcp_servers async

sync_mcp_servers(servers: Sequence[MCPServerConfig]) -> list[str]

Make configured MCP servers match servers.

If the configuration is unchanged, connected servers are left alone. If the configuration changed, existing MCP connections are closed, removed server configs are forgotten, new configs are stored, and the current server set is connected.

get_tool

get_tool(name: str) -> Tool | None

Return one registered Tool object by name.

get_tools

get_tools(tool_names: list[str]) -> list[Tool]

Return registered Tool objects filtered by name.

list_tools

list_tools() -> list[Tool]

Return all registered Tool objects.

get_openai_tool_definitions

get_openai_tool_definitions(tool_names: list[str] | None = None) -> list[dict[str, Any]]

Return OpenAI-compatible tool definitions for a model request.

tool_names=None returns all tools, [] returns none, and a non-empty list returns only the named tools in the given order.

execute_call async

execute_call(call: ToolCall, *, tool_names: list[str] | None = None, workspace: Path | str | None = None, approval_handler: ToolApprovalHandler | None = None, emitter: EventEmitter | None = None) -> ToolResult

Execute one tool call with a run-scoped workspace and optional approval.

execute_many async

execute_many(calls: list[ToolCall], *, tool_names: list[str] | None = None, workspace: Path | str | None = None, approval_handler: ToolApprovalHandler | None = None, emitter: EventEmitter | None = None) -> list[ToolResult]

Execute tool calls with a run-scoped workspace and optional approval.

close_mcp_server async

close_mcp_server(server_name: str) -> None

Close one MCP server connection and unregister its tools.

close_mcp async

close_mcp() -> None

Close all MCP server connections and unregister their tools.

close async

close() -> None

Release all resources owned by this manager.

保存工具定义,准备参数并查询可用工具。

ToolRegistry

Registry used by the agent loop to expose and execute tools.

unregister

unregister(name: str) -> None

Remove a registered tool by name if it exists.

prepare_call

prepare_call(name: str, arguments: dict[str, Any]) -> PreparedToolCall

Resolve a tool call and prepare its arguments for execution.

tool

tool(fn_or_name: Callable[..., Any] | str | None = None, *, name: str | None = None, description: str | None = None, parameters: dict[str, Any] | None = None, parallel_safe: bool = False) -> Callable[[Callable[..., Any]], Callable[..., Any]] | Callable[..., Any]

Register a function as a tool.

get_tools

get_tools(tool_names: list[str]) -> list[Tool]

Return registered tools filtered by name.

list_tools

list_tools() -> list[Tool]

Return all registered tools.

get_openai_tool_definitions

get_openai_tool_definitions(tool_names: list[str] | None = None) -> list[dict[str, Any]]

Return OpenAI-compatible tool definitions for the model request.

tool_names=None returns all tools, [] returns none, and a non-empty list returns only the named tools in the given order.

工具定义

把普通 Python 函数转换成工具。

CallableTool dataclass

Bases: Tool

A callable object exposed as an LLM-callable tool.

execute async

execute(**kwargs: Any) -> Any

Execute the wrapped function, supporting sync and async functions.

自定义工具基类,定义 schema、参数校验与执行。

Tool dataclass

Bases: ABC

Base class for LLM-callable tools.

Tools execute independently by default. Set parallel_safe=True only when the handler can safely overlap with other parallel-safe tool calls.

cast_arguments

cast_arguments(arguments: dict[str, Any]) -> dict[str, Any]

Apply safe schema-driven casts before validation.

validate_arguments

validate_arguments(arguments: dict[str, Any]) -> None

Validate arguments against the tool JSON Schema.

prepare_arguments

prepare_arguments(arguments: dict[str, Any]) -> dict[str, Any]

Cast and validate arguments before tool execution.

to_openai_tool_schema

to_openai_tool_schema() -> dict[str, Any]

Return the OpenAI-compatible tool schema.

execute abstractmethod async

execute(**kwargs: Any) -> Any

Execute the tool.

已经解析的工具调用,包含 ID、名称与参数。

ToolCall dataclass

A parsed tool call ready for registry execution.

to_openai_tool_call

to_openai_tool_call() -> dict[str, Any]

Return this call in Chat Completions assistant-message shape.

工具的内容或结构化错误。

ToolResult dataclass

Result of executing a tool call.

to_openai_tool_message

to_openai_tool_message(*, call: ToolCall | None = None) -> dict[str, Any]

Return this result in Chat Completions tool-message shape.

把模型的工具调用数据转换为 ToolCall。

parse_tool_call

parse_tool_call(raw: Any) -> ToolCall

Parse and validate a raw model tool call into a ToolCall.

执行前审批

处理器读取 call_id、name 和校验后的 arguments。

ToolApprovalRequest dataclass

One validated tool call awaiting an execution decision.

approve() 批准;reject(reason) 拒绝并把原因交给 Agent。

ToolApprovalDecision dataclass

Decision returned by a tool approval handler.

异步审批函数类型。同一批次可有多个并行审批请求;处理器异常会使当前工具调用失败。

ToolApprovalHandler module-attribute

ToolApprovalHandler = Callable[[ToolApprovalRequest], Awaitable[ToolApprovalDecision]]

相关类型

MCPServerStatus 的完整定义已集中到对应主题。