Tools¶

Agent tool system.

Provides the @tool decorator and tool-related types for registering Python functions as tools that can be called by LLMs.

agent.tools.tool(func=None, *, name=None, description=None, timeout=None, max_retries=0)[source]¶

Overloads:

func (Callable[…, Any]) → Tool
func (None), name (str | None), description (str | None), timeout (float | None), max_retries (int) → Callable[[Callable[…, Any]], Tool]

Parameters:

func (Callable[[...], Any] | None)
name (str | None)
description (str | None)
timeout (float | None)
max_retries (int)

Return type:

Tool | Callable[[Callable[[…], Any]], Tool]

Decorator to register a function as a tool.

Usage:

@tool def search(query: str) -> str:

‘’’Search for information.’’’ return f”Results for: {query}”

@tool(name=”custom_name”, timeout=30.0) def fetch_data(url: str) -> str:

‘’’Fetch data from a URL.’’’ …

Parameters:

func (Callable[..., Any] | None)
name (str | None)
description (str | None)
timeout (float | None)
max_retries (int)

Return type:

Tool | Callable[[Callable[[…], Any]], Tool]

class agent.tools.Tool(spec, function, is_async=False, timeout=None, max_retries=0)[source]¶

Bases: object

A registered tool with its specification and callable function.

Parameters:

spec (ToolSpec)
function (Callable[..., Any])
is_async (bool)
timeout (float | None)
max_retries (int)

property name: str¶: Get tool name.

async execute(arguments)[source]¶

Execute the tool with given arguments.

Parameters:: arguments (dict[str, Any])
Return type:: str

execute_sync(arguments)[source]¶

Execute the tool synchronously.

Parameters:: arguments (dict[str, Any])
Return type:: str

class agent.tools.ToolSpec(**data)[source]¶

Bases: BaseModel

Specification for a tool that can be called by an LLM.

Parameters:

data (Any)
name (str)
description (str)
parameters (dict[str, Any])
function (Any | None)
is_async (bool)

name: str¶

description: str¶

parameters: dict[str, Any]¶

function: Any | None¶

is_async: bool¶

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

to_openai_schema()[source]¶

Convert to OpenAI function calling format.

Return type:: dict[str, Any]

to_anthropic_schema()[source]¶

Convert to Anthropic tool format.

Return type:: dict[str, Any]

to_gemini_schema()[source]¶

Convert to Gemini function declaration format.

Return type:: dict[str, Any]

class agent.tools.ToolCall(**data)[source]¶

Bases: BaseModel

A tool call requested by the LLM.

Parameters:

data (Any)
id (str)
name (str)
arguments (dict[str, Any])

id: str¶

name: str¶

arguments: dict[str, Any]¶

to_dict()[source]¶

Convert to dictionary.

Return type:: dict[str, Any]

classmethod from_dict(data)[source]¶

Create from dictionary.

Parameters:: data (dict[str, Any])
Return type:: ToolCall

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class agent.tools.ToolResult(**data)[source]¶

Bases: BaseModel

Result of executing a tool.

Parameters:

data (Any)
tool_call_id (str)
name (str)
content (str)
is_error (bool)

tool_call_id: str¶

name: str¶

content: str¶

is_error: bool¶

to_dict()[source]¶

Convert to dictionary.

Return type:: dict[str, Any]

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class agent.tools.ToolRegistry[source]¶

Bases: object

Registry for managing tools.

register(tool)[source]¶

Parameters:: tool (Tool)
Return type:: None

get(name)[source]¶

Get a tool by name.

Parameters:: name (str)
Return type:: Tool | None

get_all()[source]¶

List all registered tools.

Return type:: list[Tool]

specs()[source]¶

Get specs for all registered tools.

Return type:: list[ToolSpec]

clear()[source]¶

Clear all registered tools.

Return type:: None

Agent structured output schemas.

Support for Pydantic-based schema definition and JSON Schema.

class agent.schemas.Schema(schema, *, strict=True, repair_attempts=1)[source]¶

Bases: object

Wrapper for schema definitions.

Supports both Pydantic models and raw JSON Schema dictionaries.

Parameters:

schema (type[BaseModel] | dict[str, Any])
strict (bool)
repair_attempts (int)

property json_schema: dict[str, Any]¶: Get the JSON Schema representation.

validate(data)[source]¶

Validate data against the schema.

Returns the validated/parsed object, or raises SchemaValidationError.

Parameters:: data (Any)
Return type:: Any

parse_json(text)[source]¶

Parse JSON text and validate against schema.

Handles extracting JSON from markdown code blocks.

Parameters:: text (str)
Return type:: Any

agent.schemas.schema_to_prompt(schema)[source]¶

Convert a schema to a prompt instruction.

Parameters:: schema (Schema)
Return type:: str

agent.schemas.extract_json(text)[source]¶

Extract JSON from text that may contain other content.

Handles: - Plain JSON - JSON in markdown code blocks - JSON embedded in text

Parameters:: text (str)
Return type:: dict[str, Any]

agent.schemas.repair_json(text, error)[source]¶

Attempt to repair malformed JSON.

Basic repairs: - Add missing closing braces/brackets - Fix trailing commas - Fix unquoted keys

Parameters:

text (str)
error (Exception)

Return type:

str

Tool execution loop.

Handles the cycle of LLM response -> tool execution -> continue until done.

class agent.execution.tool_loop.ToolLoopConfig(**data)[source]¶

Bases: BaseModel

Configuration for tool loop behavior.

Parameters:

data (Any)
max_iterations (int)
max_tool_calls_per_iteration (int)
timeout_per_tool (float)
parallel_tool_execution (bool)
stop_on_error (bool)

max_iterations: int¶

max_tool_calls_per_iteration: int¶

timeout_per_tool: float¶

parallel_tool_execution: bool¶

stop_on_error: bool¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class agent.execution.tool_loop.ToolLoop(tools, config=None)[source]¶

Bases: object

Manages the tool calling loop.

Executes tools requested by the LLM and continues the conversation until the LLM produces a final response without tool calls.

Parameters:

tools (list[Tool])
config (ToolLoopConfig | None)

execute_tool_calls(tool_calls)[source]¶

Execute a list of tool calls synchronously.

Parameters:: tool_calls (list[ToolCall]) – Tool calls to execute
Return type:: list[ToolResult]
Returns:: List of tool results

async execute_tool_calls_async(tool_calls)[source]¶

Execute a list of tool calls asynchronously.

Parameters:: tool_calls (list[ToolCall]) – Tool calls to execute
Return type:: list[ToolResult]
Returns:: List of tool results

build_tool_messages(response, results)[source]¶

Build messages to append to conversation after tool execution.

Parameters:

response (AgentResponse) – The LLM response containing tool calls
results (list[ToolResult]) – Results from tool execution

Return type:

list[Message]

Returns:

List of messages to append

run_loop(initial_request, run_fn)[source]¶

Run the tool loop until completion.

Parameters:

initial_request (AgentRequest) – The initial request
run_fn (Callable[[AgentRequest], AgentResponse]) – Function to call the LLM

Return type:

AgentResponse

Returns:

Final response after tool loop completion

async run_loop_async(initial_request, run_fn)[source]¶

Run the tool loop asynchronously until completion.

Parameters:

initial_request (AgentRequest) – The initial request
run_fn (Callable[[AgentRequest], Awaitable[AgentResponse]]) – Async function to call the LLM

Return type:

AgentResponse

Returns:

Final response after tool loop completion

Main execution runtime.

Orchestrates the complete request lifecycle including middleware, retries, tool loops, and structured output handling.

class agent.execution.runtime.ExecutionRuntime(provider, config, tools=None, middleware=None, retry_config=None, tool_loop_config=None)[source]¶

Bases: object

Core execution runtime for Agent.

Handles the complete lifecycle of a request: 1. Middleware preprocessing 2. Request preparation 3. Retry handling 4. Tool loop orchestration 5. Structured output parsing 6. Middleware postprocessing

Parameters:

provider (BaseProvider)
config (AgentConfig)
tools (list[Tool] | None)
middleware (MiddlewareChain | None)
retry_config (RetryConfig | None)
tool_loop_config (ToolLoopConfig | None)

run(request, schema=None)[source]¶

Execute a request synchronously.

Parameters:

request (AgentRequest) – The normalized request
schema (type[BaseModel] | dict[str, Any] | None) – Optional output schema

Return type:

AgentResponse

Returns:

Normalized response

async run_async(request, schema=None)[source]¶

Execute a request asynchronously.

Parameters:

request (AgentRequest) – The normalized request
schema (type[BaseModel] | dict[str, Any] | None) – Optional output schema

Return type:

AgentResponse

Returns:

Normalized response

stream(request)[source]¶

Execute a streaming request.

Parameters:: request (AgentRequest) – The normalized request
Return type:: StreamResponse
Returns:: Stream response iterator

async stream_async(request)[source]¶

Execute an async streaming request.

Parameters:: request (AgentRequest) – The normalized request
Return type:: AsyncStreamResponse
Returns:: Async stream response iterator