AgentixClient

AgentixClient is the primary public interface. It holds session state and exposes a simple async API.

from agentix import AgentixAgentOptions, AgentixClient

# Instantiate directly
client = AgentixClient(options=AgentixAgentOptions(...))

# Preferred: async context manager (handles connect/disconnect lifecycle)
async with AgentixClient(options) as client:
    ...

Constructor

AgentixClient(
    options: AgentixAgentOptions | None = None,
    *,
    tenant_id: str = "default",
    circuit_breaker_backend: CircuitBreakerBackend | None = None,
)

Parameter	Type	Default	Description
options	AgentixAgentOptions | None	None	Agent configuration. When None, uses AgentixAgentOptions() defaults (Anthropic provider).
tenant_id	str	"default"	Logical tenant identifier scoping the circuit breaker. Use distinct values in multi-tenant deployments.
circuit_breaker_backend	CircuitBreakerBackend | None	None	Circuit breaker state storage. Defaults to InMemoryCircuitBreakerBackend. Pass RedisCircuitBreakerBackend or AsyncRedisCircuitBreakerBackend for shared state across workers.

Circuit breaker imports: from agentix.resilience import CircuitBreakerBackend, InMemoryCircuitBreakerBackend, RedisCircuitBreakerBackend, AsyncRedisCircuitBreakerBackend

Methods

Method	Returns	Description
query(prompt, session_id=None)	AsyncGenerator[Message, None]	Run the agent. Yields all messages as they arrive.
reset_session()	None	Clear session ID; next query() starts a fresh conversation.
interrupt()	None	Signal the running query to stop (safe from another asyncio task).
await connect()	None	Prepare the client (no-op; provided for lifecycle symmetry).
await disconnect()	None	Release resources and clear session state.
set_permission_mode(mode)	None	Set permission mode for subsequent queries.
set_model(model)	None	Set the model for subsequent queries.
set_can_use_tool(callback)	None	Replace the can_use_tool permission callback.
list_tools()	list[dict]	Return registered tools available to the agent.
get_mcp_status()	list[McpServerStatus]	Return status for each connected MCP server.
on(name, callback)	None	Register a lifecycle hook callback.
rewind_files(user_message_id)	—	Rewind file state to a checkpoint. Not yet implemented — always raises NotImplementedError. Requires enable_file_checkpointing=True.

Properties

Property	Type	Description
options	AgentixAgentOptions	Agent configuration (mutable)
config	AgentConfig	Internal resolved config (model, max_tokens, cwd, etc.)
state	StateManager | StorageBackend	Session state manager
permissions	PermissionManager	Permission state
tool	decorator	Register custom tools via @client.tool(name=..., description=...)

Custom tools

@client.tool(name="MyTool", description="Does something useful.")
def my_tool(arg: str) -> str:
    return f"Result for {arg}"

Async tools are also supported:

@client.tool(name="AsyncTool", description="Async tool.")
async def async_tool(query: str) -> str:
    result = await some_async_call(query)
    return result

Lifecycle

# Option 1: async context manager (recommended)
async with AgentixClient(options) as client:
    async for msg in client.query("Hello"):
        ...

# Option 2: manual lifecycle
client = AgentixClient(options)
await client.connect()
try:
    async for msg in client.query("Hello"):
        ...
finally:
    await client.disconnect()

Anti-patterns

# ❌ Never access the internal harness
client.harness.registry.list_tools()  # AttributeError

# ✅ Use the public method
client.list_tools()

# ❌ Never create a new client per query (wastes plugin/MCP startup)
for prompt in prompts:
    async with AgentixClient(options) as c:   # expensive!
        ...

# ✅ Create once, reuse
async with AgentixClient(options) as client:
    for prompt in prompts:
        async for msg in client.query(prompt): ...