State API

Manage agent state and conversation history

Overview

The State API provides the AgentState class for managing agent execution state, conversation history, and context. Each agent maintains its own state throughout its lifecycle.

AgentState

Manages the complete state of an agent including messages, iteration count, and execution metadata.

Constructor

from strix.agents.state import AgentState

state = AgentState(
    agent_name: str = "Strix Agent",
    max_iterations: int = 300
)

agent_name · str · default: "'Strix Agent'"

Human-readable name for the agent

max_iterations · int · default: "300"

Maximum number of iterations allowed before stopping

Example:

from strix.agents.state import AgentState

state = AgentState(
    agent_name="WebScanner",
    max_iterations=100
)
print(f"Agent ID: {state.agent_id}")

Properties

Identity

agent_id · str

Unique identifier auto-generated as “agent_” + 8 random hex characters

agent_name · str · default: "'Strix Agent'"

Agent display name

parent_id · str | None · default: "None"

Parent agent ID if this is a sub-agent

Sandbox Information

sandbox_id · str | None · default: "None"

Docker container ID for the sandbox environment

sandbox_token · str | None · default: "None"

Authentication token for sandbox API

sandbox_info · dict[str, Any] | None · default: "None"

Complete sandbox information including ports and URLs

Execution State

task · str · default: "''"

Current task description

iteration · int · default: "0"

Current iteration number

max_iterations · int · default: "300"

Maximum iterations allowed

completed · bool · default: "False"

Whether execution is completed

stop_requested · bool · default: "False"

Whether stop has been requested

waiting_for_input · bool · default: "False"

Whether agent is waiting for input

llm_failed · bool · default: "False"

Whether LLM requests are failing

final_result · dict[str, Any] | None · default: "None"

Final result when completed

Data Storage

messages · list[dict[str, Any]] · default: "[]"

Conversation history (role, content, thinking_blocks)

context · dict[str, Any] · default: "{}"

Additional context storage

actions_taken · list[dict[str, Any]] · default: "[]"

Record of all actions taken with timestamps

observations · list[dict[str, Any]] · default: "[]"

Observations recorded during execution

errors · list[str] · default: "[]"

Error messages

Timestamps

start_time · str

ISO 8601 timestamp when state was created

last_updated · str

ISO 8601 timestamp of last update

waiting_start_time · datetime | None · default: "None"

When agent entered waiting state

Methods

Message Management

add_message

def add_message(
    role: str,
    content: Any,
    thinking_blocks: list[dict[str, Any]] | None = None
) -> None

Adds a message to the conversation history.

role · str · required

Message role: “user” or “assistant”

content · Any · required

Message content (string or structured content)

thinking_blocks · list[dict[str, Any]] | None

Optional thinking blocks from reasoning models

Example:

state.add_message("user", "Scan example.com for SQL injection")
state.add_message("assistant", "I'll start by checking the login form...")

get_conversation_history

def get_conversation_history() -> list[dict[str, Any]]

Returns the full conversation history.

return · list[dict[str, Any]]

List of message dictionaries

Example:

history = state.get_conversation_history()
for msg in history:
    print(f"{msg['role']}: {msg['content'][:50]}...")

Iteration Management

increment_iteration

def increment_iteration() -> None

Increments the iteration counter and updates timestamp.

Example:

state.increment_iteration()
print(f"Now at iteration {state.iteration}/{state.max_iterations}")

has_reached_max_iterations

def has_reached_max_iterations() -> bool

Checks if max iterations reached.

return · bool

True if iteration >= max_iterations

is_approaching_max_iterations

def is_approaching_max_iterations(
    threshold: float = 0.85
) -> bool

Checks if approaching max iterations.

threshold · float · default: "0.85"

Percentage threshold (0.0 to 1.0)

return · bool

True if past threshold

Example:

if state.is_approaching_max_iterations():
    print("WARNING: Approaching max iterations")

if state.has_reached_max_iterations():
    print("ERROR: Max iterations reached")
    state.set_completed({"success": False, "reason": "max_iterations"})

Execution Control

set_completed

def set_completed(
    final_result: dict[str, Any] | None = None
) -> None

Marks the agent as completed.

final_result · dict[str, Any] | None

Final result data

Example:

state.set_completed({
    "success": True,
    "vulnerabilities_found": 3,
    "severity": "medium"
})

request_stop

def request_stop() -> None

Requests the agent to stop execution.

Example:

state.request_stop()

should_stop

def should_stop() -> bool

Checks if agent should stop.

return · bool

True if stop_requested, completed, or max iterations reached

Waiting State

enter_waiting_state

def enter_waiting_state(
    llm_failed: bool = False
) -> None

Puts agent into waiting state.

llm_failed · bool · default: "False"

Whether entering wait due to LLM failure

resume_from_waiting

def resume_from_waiting(
    new_task: str | None = None
) -> None

Resumes from waiting state.

new_task · str | None

Optional new task to set

Example:

# Enter waiting state
state.enter_waiting_state()

# Later, resume
state.resume_from_waiting(new_task="Continue scanning subdomains")

is_waiting_for_input

def is_waiting_for_input() -> bool

Checks if agent is waiting for input.

return · bool

True if waiting

has_waiting_timeout

def has_waiting_timeout() -> bool

Checks if waiting timeout (600 seconds) has been reached.

return · bool

True if timeout reached

Data Recording

add_action

def add_action(
    action: dict[str, Any]
) -> None

Records an action taken.

action · dict[str, Any] · required

Action data to record

Example:

state.add_action({
    "tool": "terminal_execute",
    "command": "nmap -p 80,443 example.com"
})

add_observation

def add_observation(
    observation: dict[str, Any]
) -> None

Records an observation.

observation · dict[str, Any] · required

Observation data

Example:

state.add_observation({
    "type": "open_port",
    "port": 443,
    "service": "https"
})

add_error

def add_error(
    error: str
) -> None

Records an error message.

error · str · required

Error message

Example:

state.add_error("Connection timeout to target server")

Context Management

update_context

def update_context(
    key: str,
    value: Any
) -> None

Updates context storage.

key · str · required

Context key

value · Any · required

Context value

Example:

state.update_context("target_url", "https://example.com")
state.update_context("auth_token", "bearer_xyz123")
state.update_context("findings", [])

# Later
findings = state.context.get("findings", [])
findings.append({"type": "xss", "location": "/search"})
state.update_context("findings", findings)

Summary

get_execution_summary

def get_execution_summary() -> dict[str, Any]

Returns a comprehensive execution summary.

return · dict[str, Any]

Dictionary containing:

• agent_id, agent_name, parent_id
• sandbox_id, sandbox_info
• task, iteration, max_iterations
• completed, final_result
• start_time, last_updated
• total_actions, total_observations, total_errors
• has_errors, max_iterations_reached

Example:

summary = state.get_execution_summary()
print(f"Agent: {summary['agent_name']}")
print(f"Iterations: {summary['iteration']}/{summary['max_iterations']}")
print(f"Actions taken: {summary['total_actions']}")
print(f"Errors: {summary['total_errors']}")
print(f"Completed: {summary['completed']}")

has_empty_last_messages

def has_empty_last_messages(
    count: int = 3
) -> bool

Checks if last N messages are empty.

count · int · default: "3"

Number of messages to check

return · bool

True if last N messages are empty

Complete Example

import asyncio
from strix.agents.state import AgentState
from strix.tools import execute_tool

async def scan_target():
    # Create state
    state = AgentState(
        agent_name="SecurityScanner",
        max_iterations=50
    )
    
    # Set task
    state.task = "Scan example.com for vulnerabilities"
    state.add_message("user", state.task)
    
    # Set context
    state.update_context("target", "example.com")
    state.update_context("findings", [])
    
    # Execution loop
    while not state.should_stop():
        state.increment_iteration()
        
        # Check iteration limits
        if state.is_approaching_max_iterations():
            print("WARNING: Approaching max iterations")
        
        if state.has_reached_max_iterations():
            state.set_completed({
                "success": False,
                "reason": "max_iterations"
            })
            break
        
        # Execute some action
        state.add_message("assistant", "Running port scan...")
        
        try:
            result = await execute_tool(
                "terminal_execute",
                agent_state=state,
                command="echo 'Scanning ports...'"
            )
            
            state.add_action({
                "tool": "terminal_execute",
                "command": "port_scan"
            })
            
            state.add_observation({
                "scan_result": result
            })
            
        except Exception as e:
            state.add_error(str(e))
            state.request_stop()
            break
        
        # Simulate completion after a few iterations
        if state.iteration >= 3:
            findings = state.context.get("findings", [])
            state.set_completed({
                "success": True,
                "vulnerabilities": len(findings)
            })
    
    # Get summary
    summary = state.get_execution_summary()
    print(f"\nExecution Summary:")
    print(f"  Agent: {summary['agent_name']}")
    print(f"  Iterations: {summary['iteration']}")
    print(f"  Actions: {summary['total_actions']}")
    print(f"  Errors: {summary['total_errors']}")
    print(f"  Completed: {summary['completed']}")
    print(f"  Result: {state.final_result}")

if __name__ == "__main__":
    asyncio.run(scan_target())