Examples

Runnable recipes. The first three mirror the repo's examples/ directory (01_hello_agent.py, 02_custom_tools.py, 03_multi_agent.py, 04_web_app.py).

Hello agent

The minimal loop — one agent, one tool, in-memory persistence, streamed to stdout. See the fully annotated version in Getting started.

engine = AgentEngine(llm=AgentLLM(), tool_registry=tools, repository=repo)
async for event in engine.run("session-1", "Say hello to Alice", agent_def):
    if event.data.get("text"):
        print(event.data["text"], end="", flush=True)

Custom tools

@tool accepts async or sync functions (sync runs in a thread), and can request the execution context:

from fg_agents import ExecutionContext, ToolRegistry, tool

@tool(description="Look up an order by id", timeout_seconds=10, retry_max=1)
async def get_order(order_id: str, ctx: ExecutionContext) -> str:
    # ctx.tenant_id / ctx.user_id come from the verified session — use them to scope
    # every query. ctx.working_memory is the session scratch space.
    ctx.working_memory.add_finding(f"looked up order {order_id}")
    return await orders_api.fetch(ctx.tenant_id, order_id)

@tool(description="Today's date")
def today() -> str:                      # sync is fine — wrapped in a thread
    from datetime import date
    return date.today().isoformat()

tools = ToolRegistry()
tools.register_many([get_order, today])

The ctx parameter is auto-injected and never appears in the schema the model sees. Reference the tools by name in AgentDefinition(tools=["get_order", "today"]).

Multi-agent orchestration

Tools are registered once in a shared registry; sub-agents are registered on the Orchestrator; the brain gets no explicit tools — manage_agent and the orchestrator prompt are injected for it:

from fg_agents import AgentDefinition, AgentLLM, Orchestrator, ToolRegistry, create_repository

researcher = AgentDefinition(name="researcher", model="openai:gpt-4.1",
                             system_prompt="You research topics thoroughly.",
                             tools=["search"])
writer = AgentDefinition(name="writer", model="openai:gpt-4.1",
                         system_prompt="You write clear summaries.")
brain = AgentDefinition(name="brain", model="anthropic:claude-sonnet-4-6")

orchestrator = Orchestrator(llm=llm, tool_registry=main_tools, repository=repo)
orchestrator.register_sub_agent("researcher", researcher)
orchestrator.register_sub_agent("writer", writer)
await orchestrator.initialize()

async for event in orchestrator.run("session-1",
                                    "Research and summarize AI agent trends.", brain):
    match event.type.value:
        case "subagent.started":
            print(f"→ {event.data['agent_name']} started")
        case "subagent.completed":
            print(f"← {event.data['agent_name']} done")
        case "session.completed":
            print(event.data["final_output"])

await orchestrator.shutdown()

The brain decides when to spawn, spawn_parallel, or message its workers; each sub-agent runs in a linked child session with its own context, a nudge timer, and a hard deadline.

Full web app + SSE client

from fg_agents import AgentDefinition, ToolRegistry, create_app, tool

@tool(description="Search the knowledge base")
async def search(query: str) -> str: ...

assistant = AgentDefinition(name="assistant", model="anthropic:claude-sonnet-4-6",
                            system_prompt="You are the product assistant.",
                            tools=["search"])
tools = ToolRegistry(); tools.register_function(search)

app = create_app(
    agents={"assistant": assistant},
    tool_registry=tools,
    db_url="postgresql+asyncpg://user:pass@localhost:5432/mydb",
)
# uvicorn myapp:app

On the client, POST /api/agent/sessions/{id}/messages returns a text/event-stream; each StreamEvent arrives as an SSE message. The repo ships reference consumers in examples/frontend/ — sse-client.js (vanilla) and use-agent.ts (React hook). Runs are detached from the request, so a dropped connection re-attaches via GET /sessions/{id}/stream with Last-Event-ID.

For an existing FastAPI app, mount the router instead and plug in your auth:

from fg_agents import create_agent_router

router = create_agent_router(orchestrator, agent_definitions={"assistant": assistant},
                             authenticator=my_authenticator)
app.include_router(router)

Permission middleware

Pattern-based tool gating; ask without an approval callback fails closed:

from fg_agents import AgentEngine, PermissionMiddleware, PermissionRule

async def approve(tool_call, context) -> bool:
    # e.g. push a confirmation to the user's browser and await the answer
    return await ui_confirm(context.user_id, tool_call.tool_name)

perms = PermissionMiddleware(
    rules=[
        PermissionRule(tool_pattern="manage_*", level="auto_approve"),
        PermissionRule(tool_pattern="delete_*", level="ask", reason="destructive"),
    ],
    denied_tools=["run_shell"],
    default_level="auto_approve",
    approval_callback=approve,
)

engine = AgentEngine(llm=llm, tool_registry=tools, repository=repo,
                     middleware=[perms])

A blocked call surfaces to the model as a denied ToolResult — the agent sees why and can route around it.

Scheduler cron job

from fg_agents import AgentScheduler   # requires: pip install "fg-agents[scheduler]"

scheduler = AgentScheduler(
    orchestrator=orchestrator,
    repository=repo,
    agent_definitions={"reporter": reporter_def},
    poll_interval_seconds=30,
)

scheduler.schedule_cron("daily-digest", agent_id="reporter",
                        cron="0 7 * * 1-5",
                        message="Compile the overnight activity digest.")
scheduler.schedule_on_event("on-signup", agent_id="reporter",
                            event="user.signup",
                            message="Welcome the new user.")
await scheduler.start()
# elsewhere in the app:
await scheduler.trigger_event("user.signup", payload={...})

Schedules are stored in the repository, so they survive restarts; the poller picks up due entries every poll_interval_seconds. run_now fires a schedule immediately; pre_execute_hook / post_execute_hook are your extension points.

Resume and cancel

Cancellation is cooperative and cascades to sub-agents:

# nuclear stop — cancels the session and all child sessions
n = await orchestrator.cancel(session_id)     # returns count of cancelled runs

# over HTTP
# POST /api/agent/sessions/{id}/cancel

A cancelled or failed session is not dead — sending a new message resumes it (the engine lifts cancelled back to running, and a cancellation baseline ensures the old cancel signal can't kill the new run):

# POST /api/agent/sessions/{id}/resume     — resume waiting_input / failed
# POST /api/agent/sessions/{id}/messages   — continue the conversation
async for event in engine.run(session_id, "Pick up where you left off.", agent_def):
    ...

Because messages, session state, and agent memory all live in the repository, resume works across process restarts — the defining property of an agent built for the web rather than a laptop.