Skip to content

Agent

The Agent class is the main entry point for creating AI agents in agex. Each agent manages its own set of registered capabilities (see Registration) and can execute tasks (see Task) through a secure Python environment.

Constructor

Agent(
    primer: str | None = None,
    eval_timeout_seconds: float = 5.0,
    max_iterations: int = 10,
    name: str | None = None,
    capabilities_primer: str | None = None,
    llm: LLM | None = None,
    llm_max_retries: int = 2,
    host: Host | None = None,
    state: StateConfig | None = None,
    fs: FSConfig | None = None,
    log_high_water_tokens: int | None = None,
    log_low_water_tokens: int | None = None,
    max_memory_mb: int | None = None,
    max_open_files: int | None = None,
)

Parameters

Parameter Type Default Description
primer str \| None None Instructions that guide the agent's behavior and personality
eval_timeout_seconds float 5.0 Maximum time in seconds for agent-generated code execution
max_iterations int 10 Maximum number of think-act cycles per task
name str \| None None Unique identifier for the agent (auto-generated if not provided)
capabilities_primer str \| None None Optional curated text that replaces the default capabilities listing
llm LLM \| None None LLM client from connect_llm(). If None, uses environment defaults. See LLM Configuration.
llm_max_retries int 2 Number of times to retry a failed LLM completion
host Host \| None None Execution host from connect_host(). If None, runs locally. See Host Configuration.
state StateConfig \| None None State config from connect_state(). If None, tasks are stateless. See State Configuration.
fs FSConfig \| None connect_fs(type="virtual") FileSystem config from connect_fs(). Defaults to an in-memory Virtual Filesystem (VFS). Pass None to disable. See FileSystem Configuration.
log_high_water_tokens int \| None None Trigger event log summarization when tokens exceed this threshold
log_low_water_tokens int \| None None Target token count after summarization (defaults to 50% of high water)
max_memory_mb int \| None None Maximum memory headroom per task in MB (Unix only). See Resource Limits.
max_open_files int \| None None Maximum file descriptors per task (Unix only). See Resource Limits.

Basic Example

from agex import Agent, connect_llm, connect_state

agent = Agent(
    name="analyst",
    primer="You are an expert data analyst.",
    llm=connect_llm(provider="openai", model="gpt-4.1-nano"),
    state=connect_state(type="versioned", storage="memory"),
)

@agent.task
def analyze(data: str) -> dict:
    """Analyze the provided data."""
    pass

result = analyze("sales data for Q1")

Configuration

LLM Configuration

Configure how the agent communicates with language models:

from agex import Agent, connect_llm

# Explicit provider and model
llm = connect_llm(provider="anthropic", model="claude-haiku-4-5")
agent = Agent(llm=llm)

# Or use environment defaults
agent = Agent()  # Uses AGEX_LLM_PROVIDER, AGEX_LLM_MODEL env vars

See LLM Configuration for providers, timeouts, and advanced options.

State Configuration

Configure agent memory and persistence:

from agex import Agent, connect_state

# Versioned state with in-memory storage
agent = Agent(
    state=connect_state(type="versioned", storage="memory"),
)

# Persistent disk storage
agent = Agent(
    state=connect_state(type="versioned", storage="disk", path="/var/agex/state"),
)

See State Configuration for sessions, storage backends, and advanced options.

Host Configuration

Configure where agent tasks execute:

from agex import Agent, connect_host, connect_state

# Local execution (default)
agent = Agent()

# Remote execution
agent = Agent(
    host=connect_host(provider="http", url="http://agent-server:8000"),
    state=connect_state(type="versioned", storage="disk", path="/shared/state"),
)

See Host Configuration for remote execution and distributed deployments.

FileSystem Configuration

Configure secure, state-backed filesystem access:

from agex import Agent, connect_fs

# Enable virtual filesystem
agent = Agent(
    state=connect_state(type="versioned", storage="disk", path="/tmp/state"),
    fs=connect_fs(type="virtual"),
)

See FileSystem Configuration for virtual filesystems, file uploads, and events.

Properties

.name

Type: str

The agent's unique identifier. Auto-generated if not provided.

agent = Agent()
print(agent.name)  # "agent_abc123" (auto-generated)

named_agent = Agent(name="my_assistant")
print(named_agent.name)  # "my_assistant"

.primer

Type: str | None

The agent's behavioral instructions.

.eval_timeout_seconds

Type: float

Maximum time for a single block of agent-generated code to execute (not LLM call time).

.max_iterations

Type: int

Maximum think-act cycles per task. Raises TaskTimeout if exceeded.

Class Methods

Agent.clone_registrations(source, *, name=None, **kwargs)

Creates a new agent with copied registrations (modules, functions, classes) but independent state/fs/host.

from agex import Agent, connect_state

sandbox = Agent.clone_registrations(
    main_agent,
    name="sandbox",
    state=connect_state(type="versioned", storage="memory"),
)
sandbox.module(ui)  # Doesn't affect main_agent
Parameter Type Default Description
source Agent Required The agent to copy registrations from
name str \| None None Name for the new agent
primer str \| None None Primer string
eval_timeout_seconds float 5.0 Code execution timeout
max_iterations int 10 Max think-act cycles
llm LLM \| None None LLM configuration
host Host \| None None Execution host (defaults to Local())
state StateConfig \| None None State configuration (defaults to ephemeral)
fs FSConfig \| None Default FileSystem configuration (defaults to VirtualFS)

Returns: A new Agent with copied registrations and independent state/fs/host.

Standalone Functions

run_file_in_sandbox(agent, file_path, session, **kwargs)

Runs a file from VFS in the agent's sandbox.

from agex import run_file_in_sandbox

state = run_file_in_sandbox(sandbox, "app/main.py", session_id)
Parameter Type Default Description
agent Agent Required The agent providing the execution context
file_path str Required Path to the file in VFS
session str "default" Session identifier for state/fs access
eval_timeout_seconds float \| None None Optional timeout override

Returns: The State after execution.

Raises: FileNotFoundError if the file doesn't exist, EvalError if execution fails.

Instance Methods

.state(session: str = "default")

Returns the agent's state object for a given session. This is useful for: - Inspecting state with view(state) (see View API) - Reading event history with events(state) (see Events API) - Task cancellation (see Task Cancellation)

from agex import Agent, connect_state, view, events

agent = Agent(
    state=connect_state(type="versioned", storage="disk", path="/tmp/state"),
)

@agent.task
def my_task() -> str:
    """Do something."""
    pass

my_task()

# Inspect state
state = agent.state()  # Default session
print(view(state))

# Get events
for event in events(state):
    print(event)

# Specific session
state = agent.state(session="user_123")

Host compatibility:

Host Access
Local Full access
HTTP ❌ Not supported (state is on remote server)
Modal Full access

Capabilities Primer

By default, the agent's system message includes capabilities rendered from registrations. You can override this with curated text:

from agex import summarize_capabilities

# Generate a curated primer
primer_text = summarize_capabilities(agent, target_chars=8000)
agent.capabilities_primer = primer_text

summarize_capabilities() Helper

def summarize_capabilities(
    agent: Agent,
    target_chars: int,
    llm: LLM | None = None,
    use_cache: bool = True,
) -> str

Generates a concise, guidance-oriented primer from the agent's registrations. Results are cached at .agex/primer_cache/.

Event Log Summarization

For long-running agents, automatic summarization keeps the context window manageable:

agent = Agent(
    log_high_water_tokens=20000,  # Trigger at 20k tokens
    log_low_water_tokens=10000,   # Target 10k after summarization
)

How it works: 1. Before each LLM call, checks total event tokens 2. If exceeding high water, summarizes oldest events 3. Replaces old events with a SummaryEvent

See Events - SummaryEvent for details.

Agent Registry

agex registers all agents in a global registry for inter-agent communication. For testing, clear the registry between tests:

from agex import clear_agent_registry
import pytest

@pytest.fixture(autouse=True)
def clear_agents():
    clear_agent_registry()
    yield
    clear_agent_registry()

Advanced: Custom System Instructions

Override the built-in agex primer for specialized architectures:

# WARNING: Removes all built-in safety rails
agent = Agent(
    agex_primer_override="Custom instructions for specialized agent..."
)

Use for A/B testing system prompts or model-specific optimizations.

Resource Limits

Protect against runaway agent code with per-task resource limits:

agent = Agent(
    max_memory_mb=500,    # Allow 500MB additional memory per task
    max_open_files=256,   # Allow up to 256 file descriptors
)

Memory Limits

Memory limits use a delta-based headroom approach: the limit is set to current process memory + configured headroom. So max_memory_mb=500 means each task can allocate up to 500MB of additional memory beyond what the process was already using.

# If agent code tries to allocate too much memory:
data = bytearray(1024 * 1024 * 1024)  # 1GB allocation

# With max_memory_mb=500, this raises MemoryError
# (wrapped in EvalError when running through the sandbox)

File Descriptor Limits

Limits the number of open file descriptors to prevent resource exhaustion:

agent = Agent(max_open_files=100)

# If agent opens too many files:
files = [open(f"file{i}.txt", "w") for i in range(200)]
# Raises OSError: Too many open files

Platform Support

Platform Support
Linux Full support via RLIMIT_AS and RLIMIT_NOFILE
macOS Full support via RLIMIT_AS and RLIMIT_NOFILE
Windows Not supported (warns and continues without limits)

Process-Level Behavior

Resource limits are process-wide on Unix. For concurrent tasks in the same process, the limit applies to all tasks combined. Size your limits according to expected concurrency:

# Running 4 concurrent tasks, each needs up to 200MB
agent = Agent(max_memory_mb=800)  # 4 × 200MB headroom

For stronger isolation guarantees with per-task limits, use the Modal integration which provides containerized execution.

VFS Size Limits

For virtual filesystem size limits, see FileSystem Configuration.

Network Access Control

By default, agent code cannot make network connections. This prevents data exfiltration and unauthorized API calls. To allow specific functions to use the network, register them with network_access=True:

@agent.fn(network_access=True)
def fetch_data(url: str) -> dict:
    """Fetch data from a URL."""
    import requests
    return requests.get(url).json()

See Security - Network Access Control for details.

Next Steps