Agent

from elsai import Agent

The core class for interacting with LLMs and tools.

---

Agent.__init__

Agent(
    model=None,
    messages=None,
    tools=None,
    system_prompt=None,
    structured_output_model=None,
    callback_handler=<default>,
    conversation_manager=None,
    record_direct_tool_call=True,
    load_tools_from_directory=False,
    trace_attributes=None,
    *,
    agent_id=None,
    name=None,
    description=None,
    state=None,
    plugins=None,
    hooks=None,
    session_manager=None,
    structured_output_prompt=None,
    tool_executor=None,
    retry_strategy=<default>,
    concurrent_invocation_mode=ConcurrentInvocationMode.THROW,
)

Parameters

Name	Type	Default	Description
model	Model | str | None	BedrockModel()	LLM provider. Pass a Model instance, a Bedrock model ID string, or None for the default.
messages	list[Message] | None	[]	Pre-loaded conversation history.
tools	list | None	None	Tools available to the agent.
system_prompt	str | list[SystemContentBlock] | None	None	System instructions.
structured_output_model	type[BaseModel] | None	None	Pydantic model for typed responses.
callback_handler	Callable | None	PrintingCallbackHandler()	Event callback. Pass None for silent operation.
conversation_manager	ConversationManager | None	SlidingWindowConversationManager()	History management strategy.
record_direct_tool_call	bool	True	Record direct agent.tool.* calls in message history.
load_tools_from_directory	bool	False	Auto-load tools from ./tools/ with hot-reload.
trace_attributes	Mapping[str, AttributeValue] | None	None	Custom OpenTelemetry span attributes.
agent_id	str | None	"default"	Agent identifier for sessions and multi-agent.
name	str | None	"Elsai Agents"	Human-readable agent name.
description	str | None	None	Agent description (used in as_tool()).
state	AgentState | dict | None	AgentState()	Shared mutable state.
plugins	list[Plugin] | None	None	Plugin instances.
hooks	list[HookProvider | HookCallback] | None	None	Lifecycle event hooks.
session_manager	SessionManager | None	None	Session persistence.
structured_output_prompt	str | None	None	Custom prompt for structured output fallback.
tool_executor	ToolExecutor | None	ConcurrentToolExecutor()	Tool execution strategy.
retry_strategy	ModelRetryStrategy | None	Default strategy	Retry on transient errors.
concurrent_invocation_mode	ConcurrentInvocationMode	THROW	How to handle concurrent invocations.

---

Agent.__call__

result = agent(prompt=None, *, invocation_state=None, structured_output_model=None, structured_output_prompt=None)

Process input through the agent loop synchronously.

Parameters

Name	Type	Description
prompt	str | list[ContentBlock] | list[Message] | None	User input.
invocation_state	dict | None	Extra state passed through the event loop.
structured_output_model	type[BaseModel] | None	Override structured output model for this call.
structured_output_prompt	str | None	Override structured output prompt for this call.

Returns

AgentResult

---

Agent.invoke_async

result = await agent.invoke_async(prompt=None, *, invocation_state=None, structured_output_model=None)

Async version of __call__. Returns AgentResult.

---

Agent.stream_async

async for event in agent.stream_async(prompt=None, *, invocation_state=None):
    ...

Async iterator that yields event dicts as the agent runs.

Event keys

Key	Type	Description
data	str	Text chunk being generated
complete	bool	True when text generation is done
current_tool_use	dict	Tool call in progress (name, input)
result	AgentResult	Final result (last event)

---

Agent.cancel

agent.cancel()

Thread-safe cancellation. The agent stops at the next checkpoint and returns stop_reason="cancelled".

---

Agent.add_hook

agent.add_hook(callback, event_type=None)

Register a hook callback. Event type is inferred from the callback's type hint if not specified.

---

Agent.as_tool

tool = agent.as_tool(*, name=None, description=None, preserve_context=False)

Wrap this agent as a tool for use by another agent.

---

Agent.take_snapshot

snapshot = agent.take_snapshot(*, preset=None, include=None, exclude=None, app_data=None)

Capture current agent state as a Snapshot.

---

Agent.load_snapshot

agent.load_snapshot(snapshot)

Restore agent state from a Snapshot.

---

Agent.cleanup

agent.cleanup()

Release resources (MCP connections, file watchers). Called automatically on garbage collection.

---

Properties

Property	Type	Description
system_prompt	str | None	Current system prompt as string
system_prompt_content	list[SystemContentBlock] | None	System prompt as content blocks
tool	_ToolCaller	Direct tool invocation interface (agent.tool.tool_name(...))
tool_names	list[str]	Names of all registered tools
messages	list[Message]	Current conversation history
state	AgentState	Agent state store
model	Model	The model instance

---

AgentResult

result = agent("Hello")

result.message        # dict: the final assistant message
result.stop_reason    # str: "end_turn" | "max_tokens" | "tool_use" | "cancelled"
result.metrics        # Metrics: token counts and latency
result.state          # dict: agent state at completion
result.structured_output  # BaseModel | None: parsed structured output

---

@tool decorator

from elsai import tool

@tool
def my_tool(param: str) -> str:
    """Tool description.

    Args:
        param: Parameter description.
    """
    return result

Transforms a Python function into an agent tool. The function's name, docstring, and type hints define the tool's schema.