Agent¶

Main Agent class.

The primary interface for interacting with LLM providers.

class agent.agent.Agent(provider, model, *, api_key=None, base_url=None, timeout=120.0, max_retries=2, temperature=None, max_tokens=None, top_p=None, tools=None, middleware=None, default_system=None, **kwargs)[source]¶

Bases: object

A provider-agnostic LLM agent.

The Agent class is the main entry point for interacting with LLM providers. It provides a unified interface for text generation, streaming, structured outputs, and tool calling across multiple providers.

Example

```python from agent import Agent

agent = Agent(provider=”openai”, model=”gpt-4o”) response = agent.run(“Hello, world!”) print(response.text) ```

Parameters:

provider (str)
model (str)
api_key (str | None)
base_url (str | None)
timeout (float)
max_retries (int)
temperature (float | None)
max_tokens (int | None)
top_p (float | None)
tools (list[Tool] | None)
middleware (list[Middleware] | None)
default_system (str | None)
kwargs (Any)

property provider: str¶: Get the provider name.

property model: str¶: Get the model name.

property tools: list[Tool]¶: Get registered tools.

run(input=None, *, messages=None, system=None, temperature=None, max_tokens=None, stop=None, metadata=None)[source]¶

Execute a synchronous request.

Parameters:

input (str | None) – User input text
messages (list[Message] | None) – Optional message history
system (str | None) – System prompt (overrides default)
temperature (float | None) – Sampling temperature (overrides default)
max_tokens (int | None) – Max tokens (overrides default)
stop (list[str] | None) – Stop sequences
metadata (dict[str, Any] | None) – Request metadata

Return type:

AgentResponse

Returns:

AgentResponse with the model’s response

async run_async(input=None, *, messages=None, system=None, temperature=None, max_tokens=None, stop=None, metadata=None)[source]¶

Execute an asynchronous request.

Parameters:

input (str | None) – User input text
messages (list[Message] | None) – Optional message history
system (str | None) – System prompt (overrides default)
temperature (float | None) – Sampling temperature (overrides default)
max_tokens (int | None) – Max tokens (overrides default)
stop (list[str] | None) – Stop sequences
metadata (dict[str, Any] | None) – Request metadata

Return type:

AgentResponse

Returns:

AgentResponse with the model’s response

stream(input=None, *, messages=None, system=None, temperature=None, max_tokens=None, stop=None, metadata=None)[source]¶

Execute a streaming request.

Parameters:

input (str | None) – User input text
messages (list[Message] | None) – Optional message history
system (str | None) – System prompt (overrides default)
temperature (float | None) – Sampling temperature (overrides default)
max_tokens (int | None) – Max tokens (overrides default)
stop (list[str] | None) – Stop sequences
metadata (dict[str, Any] | None) – Request metadata

Return type:

StreamResponse

Returns:

StreamResponse iterator yielding events

async stream_async(input=None, *, messages=None, system=None, temperature=None, max_tokens=None, stop=None, metadata=None)[source]¶

Execute an async streaming request.

Parameters:

input (str | None) – User input text
messages (list[Message] | None) – Optional message history
system (str | None) – System prompt (overrides default)
temperature (float | None) – Sampling temperature (overrides default)
max_tokens (int | None) – Max tokens (overrides default)
stop (list[str] | None) – Stop sequences
metadata (dict[str, Any] | None) – Request metadata

Return type:

AsyncStreamResponse

Returns:

AsyncStreamResponse iterator yielding events

json(input=None, *, schema, messages=None, system=None, temperature=None, max_tokens=None, metadata=None)[source]¶

Execute a request expecting structured JSON output.

Parameters:

input (str | None) – User input text
schema (type[BaseModel] | dict[str, Any]) – Pydantic model or JSON schema for output
messages (list[Message] | None) – Optional message history
system (str | None) – System prompt (overrides default)
temperature (float | None) – Sampling temperature (overrides default)
max_tokens (int | None) – Max tokens (overrides default)
metadata (dict[str, Any] | None) – Request metadata

Return type:

AgentResponse

Returns:

AgentResponse with parsed output in response.output

async json_async(input=None, *, schema, messages=None, system=None, temperature=None, max_tokens=None, metadata=None)[source]¶

Execute an async request expecting structured JSON output.

Parameters:

input (str | None) – User input text
schema (type[BaseModel] | dict[str, Any]) – Pydantic model or JSON schema for output
messages (list[Message] | None) – Optional message history
system (str | None) – System prompt (overrides default)
temperature (float | None) – Sampling temperature (overrides default)
max_tokens (int | None) – Max tokens (overrides default)
metadata (dict[str, Any] | None) – Request metadata

Return type:

AgentResponse

Returns:

AgentResponse with parsed output in response.output

session(session_id=None, system=None)[source]¶

Create a new session for multi-turn conversation.

Parameters:

session_id (str | None) – Optional session identifier
system (str | None) – System prompt for this session

Return type:

Session

Returns:

Session instance

with_config(**kwargs)[source]¶

Create a new Agent with modified configuration.

Parameters:: **kwargs (Any) – Configuration overrides
Return type:: Agent
Returns:: New Agent instance with modified config

Agent configuration management.

Handles API keys, environment variables, and configuration loading.

agent.config.get_api_key(provider, api_key=None)[source]¶

Get API key for a provider.

Parameters:

provider (str)
api_key (str | None)

Return type:

str | None

agent.config.get_base_url(provider, base_url=None)[source]¶

Get base URL for a provider.

Parameters:

provider (str)
base_url (str | None)

Return type:

str | None

agent.config.resolve_model(model)[source]¶

Resolve model alias to full model name.

Parameters:: model (str)
Return type:: str

agent.config.estimate_cost(model, prompt_tokens, completion_tokens)[source]¶

Estimate cost for a request.

Parameters:

model (str)
prompt_tokens (int)
completion_tokens (int)

Return type:

float | None

class agent.config.AgentConfig(**data)[source]¶

Bases: BaseModel

Configuration for an Agent instance.

Parameters:

data (Any)
provider (str)
model (str)
api_key (str | None)
base_url (str | None)
timeout (float)
max_retries (int)
temperature (float | None)
max_tokens (int | None)
top_p (float | None)
default_system (str | None)
extra (dict[str, Any])

provider: str¶

model: str¶

api_key: str | None¶

base_url: str | None¶

timeout: float¶

max_retries: int¶

temperature: float | None¶

max_tokens: int | None¶

top_p: float | None¶

default_system: str | None¶

extra: dict[str, Any]¶

resolve_config()[source]¶

Resolve model alias and get API key/base URL from environment.

Return type:: AgentConfig

with_overrides(**kwargs)[source]¶

Create a new config with overrides.

Parameters:: kwargs (Any)
Return type:: AgentConfig

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

Agent response model.

Normalized response object returned by all providers.

class agent.response.Usage(**data)[source]¶

Bases: BaseModel

Token usage information.

Parameters:

data (Any)
prompt_tokens (int)
completion_tokens (int)
total_tokens (int)

prompt_tokens: int¶

completion_tokens: int¶

total_tokens: int¶

classmethod from_dict(data)[source]¶

Create Usage from a dictionary.

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

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

class agent.response.AgentResponse(**data)[source]¶

Bases: BaseModel

Normalized response from any provider.

Parameters:

data (Any)
text (str | None)
content (list[Any])
output (Any)
provider (str)
model (str)
usage (Usage | None)
stop_reason (str | None)
tool_calls (list[ToolCall])
raw (Any)
latency_ms (float | None)
cost_estimate (float | None)
request_id (str | None)

text: str | None¶

content: list[Any]¶

output: Any¶

provider: str¶

model: str¶

usage: Usage | None¶

stop_reason: str | None¶

tool_calls: list[ToolCall]¶

raw: Any¶

latency_ms: float | None¶

cost_estimate: float | None¶

request_id: str | None¶

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

property has_tool_calls: bool¶: Check if response contains tool calls.

to_dict()[source]¶

Convert response to a dictionary.

Return type:: dict[str, Any]

Agent message types and request model.

These are the normalized internal representations used across all providers.

class agent.messages.ContentPart(**data)[source]¶

Bases: BaseModel

A part of message content (text, image, etc.).

Parameters:

data (Any)
type (Literal['text', 'image', 'image_url'])
text (str | None)
image_url (str | None)
image_data (bytes | None)
media_type (str | None)

type: Literal['text', 'image', 'image_url']¶

text: str | None¶

image_url: str | None¶

image_data: bytes | None¶

media_type: str | None¶

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

classmethod text_part(text)[source]¶

Create a text content part.

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

classmethod image_url_part(url)[source]¶

Create an image URL content part.

Parameters:: url (str)
Return type:: ContentPart

classmethod image_data_part(data, media_type='image/png')[source]¶

Create an image data content part.

Parameters:

data (bytes)
media_type (str)

Return type:

ContentPart

class agent.messages.Message(**data)[source]¶

Bases: BaseModel

A normalized message in a conversation.

Parameters:

data (Any)
role (Literal['system', 'user', 'assistant', 'tool'])
content (str | list[ContentPart])
name (str | None)
tool_call_id (str | None)
tool_calls (list[dict[str, Any]] | None)

role: Literal['system', 'user', 'assistant', 'tool']¶

content: str | list[ContentPart]¶

name: str | None¶

tool_call_id: str | None¶

tool_calls: list[dict[str, Any]] | None¶

classmethod system(content)[source]¶

Create a system message.

Parameters:: content (str)
Return type:: Message

classmethod user(content)[source]¶

Create a user message.

Parameters:: content (str | list[ContentPart])
Return type:: Message

classmethod assistant(content=None, tool_calls=None)[source]¶

Create an assistant message.

Parameters:

content (str | None)
tool_calls (list[dict[str, Any]] | None)

Return type:

Message

classmethod tool(content, tool_call_id, name=None)[source]¶

Create a tool result message.

Parameters:

content (str)
tool_call_id (str)
name (str | None)

Return type:

Message

property text: str¶: Get the text content of the message.

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

class agent.messages.AgentRequest(**data)[source]¶

Bases: BaseModel

A normalized request to be sent to a provider.

Parameters:

data (Any)
input (str | None)
messages (list[Message])
system (str | None)
tools (list[Any])
output_schema (dict[str, Any] | None)
temperature (float | None)
max_tokens (int | None)
top_p (float | None)
stop (list[str] | None)
metadata (dict[str, Any])
session_id (str | None)

input: str | None¶

messages: list[Message]¶

system: str | None¶

tools: list[Any]¶

output_schema: dict[str, Any] | None¶

temperature: float | None¶

max_tokens: int | None¶

top_p: float | None¶

stop: list[str] | None¶

metadata: dict[str, Any]¶

session_id: str | None¶

property schema: dict[str, Any] | None¶: Alias for output_schema for backwards compatibility.

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

to_messages()[source]¶

Convert request to a list of messages.

Return type:: list[Message]