🎉 ai-infra v1.0 is here — Production-ready AI/LLM infrastructure
What's new
nfrax logonfrax

Infrastructure that just works. Ship products, not boilerplate.

Frameworks

  • svc-infra
  • ai-infra
  • fin-infra
  • robo-infra

Resources

  • Getting Started
  • What's New
  • Contributing

Community

  • GitHub

© 2026 nfrax. All rights reserved.

nfrax logonfrax
Start HereWhat's New
GitHub
ai-infra / API Reference

Agent

from ai_infra.llm import Agent
View source
ai_infra.llm
Extends:BaseLLM

Agent-oriented interface (tool calling, streaming updates, fallbacks). The Agent class provides a simple API for running LLM agents with tools. Tools can be plain Python functions, LangChain tools, or MCP tools. Example - Basic usage:

python
def get_weather(city: str) -> str:
        '''Get weather for a city.'''
        return f"Weather in {city}: Sunny, 72°F"

    # Simple usage with tools
    agent = Agent(tools=[get_weather])
    result = agent.run("What's the weather in NYC?")

Example - With session memory (conversations persist):

python
from ai_infra.llm.session import memory

    agent = Agent(tools=[...], session=memory())

    # Conversation 1 - remembered
    agent.run("I'm Bob", session_id="user-123")
    agent.run("What's my name?", session_id="user-123")  # Knows "Bob"

    # Different session - fresh start
    agent.run("What's my name?", session_id="user-456")  # Doesn't know

Example - Pause and resume (HITL):

python
from ai_infra.llm.session import memory

    agent = Agent(
        tools=[dangerous_tool],
        session=memory(),
        pause_before=["dangerous_tool"],  # Pause before this tool
    )

    result = agent.run("Delete file.txt", session_id="task-1")

    if result.paused:
        # Show user what's pending, get approval
        print(result.pending_action)

        # Resume with decision
        result = agent.resume(session_id="task-1", approved=True)

Example - Production with Postgres:

python
from ai_infra.llm.session import postgres

    agent = Agent(
        tools=[...],
        session=postgres("postgresql://..."),
    )
    # Sessions persist across restarts

Example - Human approval (sync, per-request):

python
agent = Agent(
        tools=[dangerous_tool],
        require_approval=True,  # Console prompt for approval
    )

Example - DeepAgents mode (autonomous multi-step tasks):

python
from ai_infra.llm import Agent
    from ai_infra.llm.session import memory

    # Define specialized agents
    researcher = Agent(
        name="researcher",
        description="Searches and analyzes code",
        system="You are a code research assistant.",
        tools=[search_codebase],
    )

    writer = Agent(
        name="writer",
        description="Writes and edits documentation",
        system="You are a technical writer.",
    )

    # Create a deep agent that can delegate to subagents
    agent = Agent(
        deep=True,
        session=memory(),
        subagents=[researcher, writer],  # Agents auto-convert to subagents
    )

    # The agent can now autonomously:
    # - Read/write/edit files
    # - Execute shell commands
    # - Delegate to subagents
    # - Maintain todo lists
    result = agent.run("Refactor the auth module to use JWT tokens")
Constructor
Agent(tools: list[Any] | None = None, provider: str | None = None, model_name: str | None = None, name: str | None = None, description: str | None = None, system: str | None = None, callbacks: Callbacks | CallbackManager | None = None, on_tool_error: Literal['return_error', 'retry', 'abort'] = 'return_error', tool_timeout: float | None = None, validate_tool_results: bool = False, max_tool_retries: int = 1, require_approval: bool | list[str] | Callable[[str, dict[str, Any]], bool] = False, approval_handler: ApprovalHandler | AsyncApprovalHandler | None = None, session: SessionStorage | None = None, pause_before: list[str] | None = None, pause_after: list[str] | None = None, deep: bool = False, subagents: list[Agent | SubAgent] | None = None, middleware: Sequence[AgentMiddleware] | None = None, response_format: Any | None = None, context_schema: type[Any] | None = None, use_longterm_memory: bool = False, workspace: str | Path | Workspace | None = None, recursion_limit: int = 50, model_kwargs = {})
ParameterTypeDefaultDescription
toolslist[Any] |NoneNoneList of tools (functions, LangChain tools, or MCP tools)
providerstr|NoneNoneLLM provider (auto-detected if None)
model_namestr|NoneNoneModel name (uses provider default if None) Agent Identity (for use as subagent):
namestr|NoneNoneAgent name (required when used as a subagent)
descriptionstr|NoneNoneWhat this agent does (used by parent to decide delegation)
systemstr|NoneNoneSystem prompt / instructions for this agent Callbacks (observability):
callbacksCallbacks | CallbackManager |NoneNoneCallback handler(s) for observing agent events. Receives events for LLM calls (start, end, error, tokens) and tool executions (start, end, error). Can be a single Callbacks instance or a CallbackManager.
on_tool_errorLiteral['return_error', 'retry', 'abort']'return_error'How to handle tool execution errors:
tool_timeoutfloat|NoneNoneTimeout in seconds per tool call (None = no timeout)
validate_tool_resultsboolFalseValidate tool results match return type annotations
max_tool_retriesint1Max retry attempts when on_tool_error="retry" (default 1) Human Approval:
require_approvalbool|list[str] |Callable[[str, dict[str, Any]], bool]FalseTools that require human approval:
approval_handlerApprovalHandler | AsyncApprovalHandler |NoneNoneCustom approval handler function: - If None and require_approval is True, uses console prompts - Can be sync or async function taking ApprovalRequest Session & Persistence:
sessionSessionStorage |NoneNoneSession storage backend for conversation memory and pause/resume. Use memory() for development, postgres() for production.
pause_beforelist[str] |NoneNoneTool names to pause before executing (requires session). The agent will return a SessionResult with paused=True.
pause_afterlist[str] |NoneNoneTool names to pause after executing (requires session). DeepAgents Mode (autonomous multi-step tasks):
deepboolFalseEnable DeepAgents mode for autonomous task execution. When True, the agent has built-in tools for file operations (ls, read_file, write_file, edit_file, glob, grep, execute), todo management, and subagent orchestration.
subagentslist[Agent | SubAgent] |NoneNoneList of agents for delegation. Can be Agent instances (automatically converted) or SubAgent dicts. Agent instances must have name and description set.
middlewareSequence[AgentMiddleware] |NoneNoneAdditional middleware to apply to the deep agent.
response_formatAny|NoneNoneStructured output format for agent responses.
context_schematype[Any] |NoneNoneSchema for the deep agent context.
use_longterm_memoryboolFalseEnable long-term memory (requires session with store). Workspace Configuration:
workspacestr| Path | Workspace |NoneNoneWorkspace configuration for file operations. Can be:
recursion_limitint50Maximum number of agent iterations (default: 50). Prevents infinite loops when agent keeps calling tools without making progress. This is a critical safety measure to prevent runaway token costs. Raise only if you have monitoring in place.
model_kwargsAny{}—

Methods

On This Page

Constructoragentaresumeasyncarunasyncarun_agentasyncarun_agent_streamasyncarun_with_fallbacksasyncastreamasyncastream_agent_tokensasyncfrom_personaclassmethodresumerunrun_agentrun_with_fallbacks