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
Maximum number of iterations allowed before stopping
from strix.agents.state import AgentState
state = AgentState(
agent_name="WebScanner",
max_iterations=100
)
print(f"Agent ID: {state.agent_id}")
Properties
Identity
Unique identifier auto-generated as “agent_” + 8 random hex characters
agent_name
str
default:"'Strix Agent'"
Agent display name
Parent agent ID if this is a sub-agent
Docker container ID for the sandbox environment
Authentication token for sandbox API
sandbox_info
dict[str, Any] | None
default:"None"
Complete sandbox information including ports and URLs
Execution State
Maximum iterations allowed
Whether execution is completed
Whether stop has been requested
Whether agent is waiting for input
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
Timestamps
ISO 8601 timestamp when state was created
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
Message role: “user” or “assistant”
Message content (string or structured content)
thinking_blocks
list[dict[str, Any]] | None
Optional thinking blocks from reasoning models
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]]
List of message dictionaries
history = state.get_conversation_history()
for msg in history:
print(f"{msg['role']}: {msg['content'][:50]}...")
Iteration Management
increment_iteration
def increment_iteration() -> None
state.increment_iteration()
print(f"Now at iteration {state.iteration}/{state.max_iterations}")
has_reached_max_iterations
def has_reached_max_iterations() -> bool
True if iteration >= max_iterations
is_approaching_max_iterations
def is_approaching_max_iterations(
threshold: float = 0.85
) -> bool
Percentage threshold (0.0 to 1.0)
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
state.set_completed({
"success": True,
"vulnerabilities_found": 3,
"severity": "medium"
})
request_stop
def request_stop() -> None
should_stop
def should_stop() -> bool
True if stop_requested, completed, or max iterations reached
Waiting State
enter_waiting_state
def enter_waiting_state(
llm_failed: bool = False
) -> None
Whether entering wait due to LLM failure
resume_from_waiting
def resume_from_waiting(
new_task: str | None = None
) -> None
# Enter waiting state
state.enter_waiting_state()
# Later, resume
state.resume_from_waiting(new_task="Continue scanning subdomains")
def is_waiting_for_input() -> bool
has_waiting_timeout
def has_waiting_timeout() -> bool
Data Recording
add_action
def add_action(
action: dict[str, Any]
) -> None
state.add_action({
"tool": "terminal_execute",
"command": "nmap -p 80,443 example.com"
})
add_observation
def add_observation(
observation: dict[str, Any]
) -> None
state.add_observation({
"type": "open_port",
"port": 443,
"service": "https"
})
add_error
def add_error(
error: str
) -> None
state.add_error("Connection timeout to target server")
Context Management
update_context
def update_context(
key: str,
value: Any
) -> None
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]
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
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
Number of messages to check
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())