API reference

The headline public symbols, all importable from fg_agents. This covers the surface most apps touch; the full export list is in src/fg_agents/__init__.py.

• One-shot: ask / stream
• Model auto-detection: resolve_default_model
• Agent (facade)
• AgentRunResult / AgentRunError
• AgentDefinition
• AgentEngine
• AgentLLM
• Tools: @tool and ToolRegistry
• Persistence: create_repository
• Web layer: create_app
• Streaming: StreamEvent / EventType
• Middleware

One-shot: ask / stream

Free functions for one-shot use — no client object. Each constructs an ephemeral Agent, runs the prompt, and closes it.

from fg_agents import ask

print(await ask("What's 2+2?"))

async def ask(
    prompt: str,
    *,
    model: str | None = None,        # omit to auto-detect from the environment
    tools: list[Callable | RegisteredTool] | None = None,
    system_prompt: str = "",
    memory: str | BaseRepository = "memory",
    **kwargs,                        # anything else Agent accepts
) -> AgentRunResult

str() of the result is the answer text, so print(await ask(...)) prints just the answer.

stream(prompt, ...) takes the same arguments and yields StreamEvents as they happen; the ephemeral agent is closed when the stream is exhausted or the generator's aclose() runs:

from fg_agents import stream

async for event in stream("Tell me a story"):
    print(event.type, event.data)

If you might exit early (break/return mid-iteration), wrap the stream in contextlib.aclosing so cleanup runs deterministically instead of waiting for garbage collection:

import contextlib

async with contextlib.aclosing(stream("Tell me a story")) as events:
    async for event in events:
        if enough(event):
            break

The ladder: ask() for one-shots → Agent for conversations → AgentEngine and friends for full control.

Model auto-detection

resolve_default_model() -> str picks a provider-qualified default model from the environment. It runs whenever ask(), stream(), or Agent() is called without a model. resolve_default_model_async() is its async twin — same order and error, but the Ollama probe never blocks the event loop; it's what ask()/stream() use internally. The probe result is cached briefly so repeated constructions don't re-pay the connect timeout. Detection order:

1. ANTHROPIC_API_KEY → anthropic:claude-sonnet-4-6
2. OPENAI_API_KEY → openai:gpt-5.2
3. GOOGLE_API_KEY or GEMINI_API_KEY → google:gemini-2.5-flash
4. A local Ollama server on localhost:11434 → ollama:qwen3:8b

If none apply it raises ModelDetectionError (importable from fg_agents) with instructions. An explicit model= always wins — detection never overrides a caller's choice.

Agent (facade)

One object that wires AgentLLM + ToolRegistry + a repository + AgentEngine. The fastest way to run an agent; graduate to the low-level objects at any point.

from fg_agents import Agent

agent = Agent(model="anthropic:claude-sonnet-4-6", tools=[my_func])
result = await agent.run("hello")

Constructor

Agent(
    model: str | None = None,                     # provider-qualified, e.g. "openai:gpt-5.2"; None = auto-detect
    *,
    tools: list[Callable | RegisteredTool] | None = None,
    system_prompt: str = "",
    name: str = "agent",
    memory: str | BaseRepository = "memory",      # "memory" | "sqlite" | "sqlite:<path>" | "postgres:<url>" | repository
    llm: AgentLLM | None = None,                  # pre-built client (e.g. with api_keys)
    api_keys: dict[str, str] | None = None,       # provider -> key; ignored when llm= is given
    session_id: str | None = None,                # default conversation id; fresh per instance otherwise
    **agent_kwargs,                               # extra AgentDefinition fields: max_turns, temperature, fallback_models, ...
)

tools accepts @tool-decorated functions, plain callables (auto-wrapped — schema derived from the signature, description from the docstring), or RegisteredTool instances. Duplicate tool names raise ValueError.

The repository is initialized lazily on the first run()/stream() call — forgetting to initialize is never an error. (The low-level tier requires an explicit await repo.initialize().)

Methods

Agent is also an async context manager — async with Agent(...) as agent: closes it on exit:

async with Agent(tools=[greet]) as agent:
    result = await agent.run("hello")

Failure contract

run() converts any failure — an engine-emitted error event or an exception raised mid-run — into a single AgentRunError carrying the session_id and the partial events collected so far; the original exception (when there is one) is chained as __cause__.

stream() is lower-level by design: engine failures surface either as error events in the stream or as raw exceptions from the iterator — handle both, or use run() for the uniform contract.

Graduation properties

agent.engine, agent.llm, agent.tools (the ToolRegistry), agent.repository, agent.definition, agent.session_id — the exact objects the facade built, for when you outgrow it.

AgentRunResult

Frozen dataclass returned by Agent.run:

AgentRunError

AgentRunError(message, session_id, events) — subclass of AgentFrameworkError (the root of the framework's exception hierarchy). Attributes: .session_id, .events (partial events up to the failure), .__cause__ (original exception, if any).

AgentDefinition

Declarative agent configuration (Pydantic model). Key fields:

AgentDefinition(
    name="assistant",
    model="anthropic:claude-sonnet-4-6",   # required — no default model
    system_prompt="...",
    tools=["search", "greet"],             # names registered in the ToolRegistry
    # plus: max_turns, temperature, fallback_models, sub_agents, skills, ...
)

See fg_agents.core.types.AgentDefinition for the full field list.

AgentEngine

The ReAct loop: model → tools → model → ..., emitting StreamEvents at each step. Runs are serialized per session (per-session lock) and support cooperative cancellation.

AgentEngine(
    llm: AgentLLM,
    tool_registry: ToolRegistry,
    repository: BaseRepository,
    skills_manager: SkillsManager | None = None,
    middleware: list[BaseMiddleware] | None = None,
)

async for event in engine.run(
    session_id: str,
    user_message: str,
    agent_def: AgentDefinition,
    metadata: dict | None = None,
    variables: dict | None = None,
):
    ...

run() creates or resumes the session, appends the user message, and runs the loop until completion or max_turns. The repository must be initialized (await repo.initialize()) before the first run.

AgentLLM

Provider-agnostic LLM client with streaming and tool-use support. Native Anthropic/OpenAI/Google clients plus 20+ OpenAI-compatible providers (including ollama: for local models). Provider clients are created lazily.

AgentLLM(
    api_keys: dict[str, str] | None = None,        # provider -> key; falls back to {PROVIDER}_API_KEY env vars
    custom_providers: dict[str, str] | None = None # provider name -> base URL, for providers not built in
)

Model ids are always provider-qualified: "anthropic:claude-sonnet-4-6", "openai:gpt-5.2", "ollama:qwen3:8b".

Tools

@tool decorator

from fg_agents import tool

@tool(name="search_db", description="Search the case database")
async def search_db(query: str, limit: int = 10) -> dict:
    ...

tool(
    name: str | None = None,              # default: function name
    description: str | None = None,       # default: first docstring line
    tool_type: ToolType = ToolType.FUNCTION,
    permission_level: str = "auto_approve",
    timeout_seconds: int = 30,
    retry_max: int = 0,
    tags: list[str] | None = None,
    json_schema: dict | None = None,      # explicit schema; otherwise derived from the signature
)

Sync functions are wrapped to run in a thread; async functions run natively. The decorated function gains a .tool_definition attribute (a RegisteredTool). A parameter named ctx/context typed ExecutionContext is injected by the framework and hidden from the JSON schema.

ToolRegistry

Central tool store handed to the engine or create_app.

Persistence

from fg_agents import create_repository

repo = create_repository("memory")                                   # tests / throwaway
repo = create_repository("sqlite", db_path="agents.db")              # local dev (needs fg-agents[sqlite])
repo = create_repository("postgres", db_url="postgresql+asyncpg://…")# production (needs fg-agents[postgres])
await repo.initialize()   # required before use (the Agent facade does this for you)

All backends implement BaseRepository (async session/message CRUD). Details and dependency requirements: persistence.md.

Web layer

Requires the web extra (python -m pip install --upgrade "fg-agents[web]"); without it, create_app/create_agent_router raise ImportError with the install hint.

from fg_agents import create_app

app = create_app(
    agents: dict[str, AgentDefinition],       # named agents exposed by the API
    db_url: str | None = None,                # postgres URL for production; SQLite file if None
    api_keys: dict[str, str] | None = None,
    tool_registry: ToolRegistry | None = None,
    cors_origins: list[str] | None = None,
    prefix: str = "/api/agent",
    title: str = "Fareground Agent API",
    admin_only: bool = True,                  # POST /tools and /agents return 403 when True
    **fastapi_kwargs,
) -> FastAPI

Returns a FastAPI app with sessions, messages, and SSE streaming wired; persistence is initialized in the app lifespan. For mounting into an existing app, use create_agent_router(...) with your own authenticator for multi-tenant isolation (see fg_agents/api/router.py).

Routes (under prefix):

POST /sessions                    -> { "session_id": "..." }
POST /sessions/{id}/messages      -> text/event-stream (SSE)

Streaming

StreamEvent (Pydantic model) is the single event shape flowing Engine → iterator → SSE → frontend:

event.to_sse() renders the SSE wire frame; event.to_dict() a plain dict. Every event type and its payload: streaming.md.

Middleware

Hooks into the agent loop, passed to AgentEngine(middleware=[...]). Implement the Middleware protocol or subclass BaseMiddleware (pass-through defaults) and override what you need:

Middleware runs in registration order. Shipped implementations: AuditMiddleware, TokenTrackingMiddleware, PermissionMiddleware, RateLimitMiddleware, LoopGuardMiddleware, ContextCompressionMiddleware.

Streaming: the SSE wire format

This is the contract between the framework and your frontend. Every event the engine emits is a StreamEvent (see src/fg_agents/streaming/events.py), serialized as one Server-Sent Event frame.

The wire frame

id: <event uuid>
event: <event type, e.g. llm.text_delta>
data: <full event as JSON>

The data: JSON always has the same envelope:

{
  "id": "6f6c…",
  "type": "llm.text_delta",
  "session_id": "session-1",
  "data": { "text": "Hel" },
  "timestamp": "2026-08-10T17:02:11.412Z",
  "turn_number": 1
}

• id — unique per event; doubles as the SSE id: for Last-Event-ID reconnection (GET /sessions/{id}/stream replays and skips past it).
• type — one of the event types below; also the SSE event: name, so EventSource.addEventListener("llm.text_delta", …) works directly.
• data — the per-type payload (tables below).
• turn_number — the ReAct turn, null for session-level events.

The stream also carries periodic heartbeat comments to keep proxies from closing idle connections; SSE clients ignore them automatically.

Event types

Listed in the order a typical run emits them.

A minimal frontend loop

// POST /api/agent/sessions            -> { session_id }
// POST /api/agent/sessions/{id}/messages  (SSE response)
let reply = "";
for await (const event of sseEvents(response)) {   // your SSE parser
  const e = JSON.parse(event.data);
  switch (e.type) {
    case "llm.text_delta":    reply += e.data.text; render(reply); break;
    case "tool.executing":    showSpinner(e.data.tool_name); break;
    case "session.completed": render(e.data.final_output); break;
    case "error":             showError(e.data.message); break;
  }
}

Ready-made clients live in examples/frontend/: sse-client.js (vanilla) and use-agent.ts (React hook).

Consuming events in Python

The same StreamEvent objects come out of Agent.stream(...) and AgentEngine.run(...) directly — no SSE involved:

async for event in agent.stream("hello"):
    if event.type == EventType.LLM_TEXT_DELTA:
        print(event.data["text"], end="")

event.to_sse() produces the wire frame above; event.to_dict() the JSON envelope as a dict.

Persistence

Sessions and message history are stored through one async interface, BaseRepository, with three interchangeable backends. Pick by environment:

Choosing a backend

Agent facade (memory= argument)

Agent(model=..., memory="memory")                                # default
Agent(model=..., memory="sqlite")                                # ./fg_agents.db
Agent(model=..., memory="sqlite:./my_agent.db")                  # explicit path
Agent(model=..., memory="postgres:postgresql+asyncpg://user:pass@host/db")
Agent(model=..., memory=my_repository)                           # any BaseRepository instance

The postgres form requires the URL after the colon; a bare "postgres" raises ValueError.

Low-level (create_repository)

from fg_agents import create_repository

repo = create_repository("memory")
repo = create_repository("sqlite", db_path="agents.db")     # default path: "fg_agents.db"
repo = create_repository("postgres", db_url="postgresql+asyncpg://user:pass@host/db", echo=False)

Web app (create_app(db_url=...))

create_app picks the backend from the URL string:

• db_url=None → SQLite file fg_agents.db (dev default — needs fg-agents[sqlite] installed; without aiosqlite, create_app() raises ImportError immediately with the fix — install the extra or pass db_url="memory")
• URL containing postgres/postgresql → PostgreSQL
• URL containing sqlite → SQLite at the path after ///
• "memory" → in-memory
• anything else → ValueError (typos fail loudly instead of silently losing persistence)

Initialization semantics

• Agent facade — lazy. The repository is initialized automatically on the first run()/stream() call (exactly once, concurrency-safe). Forgetting to initialize is never an error. await agent.close() closes it.
• Low-level — explicit. You must await repo.initialize() before handing it to AgentEngine, and await repo.close() on shutdown. Skipping initialize() is a bug (tables/connections won't exist).
• create_app — managed. Initialization and shutdown are wired into the FastAPI lifespan; you do nothing.

initialize() creates schema (tables) on SQL backends, so first-run against a fresh database needs no separate migration step. PostgreSQL URLs must use the async driver: postgresql+asyncpg://....

Notes

• Importing SQLiteRepository without aiosqlite installed gives a stub that raises ImportError with the install hint when called.
• All backends implement the same async CRUD surface for sessions and messages (fg_agents/persistence/base.py), so swapping backends is a one-line change.

Web integration

A local service

Install the web extra and a model provider:

python -m pip install --upgrade "fg-agents[web,anthropic]"

Save as app.py:

import os
from fg_agents import AgentDefinition, create_app

assistant = AgentDefinition(
    name="assistant",
    model=os.environ["AGENT_MODEL"],
    system_prompt="Help the user reason about inventory decisions.",
    tools=[],
    max_turns=6,
)
app = create_app(agents={"assistant": assistant}, db_url="memory")

uvicorn app:app --host 127.0.0.1

This is a local, ephemeral demonstration. db_url="memory" avoids a database dependency. Omitting db_url selects a local SQLite file and requires the sqlite extra.

HTTP and streaming

The default prefix is /api/agent. Create a session with POST /sessions, then send messages to POST /sessions/{id}/messages; responses stream as Server-Sent Events. The generated FastAPI OpenAPI schema documents request bodies for the configured application.

Read streaming before writing a frontend parser. A network chunk is not necessarily a complete SSE event. Handle reconnects and errors deliberately.

Authentication for an exposed service

The default router uses unauthenticated, shared-tenant access. admin_only=True restricts certain registration endpoints; it is not a replacement for request authentication.

For an exposed or multi-tenant application, mount create_agent_router(..., authenticator=...) in your own FastAPI app. An authenticator validates a request and returns AuthContext from fg_agents.api.auth, or raises HTTP 401. Derive the tenant from verified credentials, not a user-supplied request body.

HeaderAuthenticator is for a trusted proxy that strips incoming identity headers and sets verified ones. It is not safe to trust those headers from arbitrary clients.

Application lifecycle

Initialize persistence before handling requests and close it during shutdown. Configure CORS for the actual frontend origins, enforce request limits, and ensure session access follows your tenant boundary. See production integration and persistence.

Production integration

The SDK supplies execution components. Your application supplies the trust boundary and operational policy.

Before exposing an agent

• Authenticate users and derive tenant identity from verified credentials.
• Register only the tools each role needs; authorize sensitive actions inside handlers.
• Set model, turn, timeout and concurrency limits explicitly.
• Choose durable persistence and test restart behavior.
• Record failures and usage without retaining secrets in logs.
• Test your actual providers, including malformed arguments, truncation and interruption.

External side effects

A tool that sends an order, changes a record or makes a payment needs application-level idempotency. A timeout may leave an external request in progress. Store an operation ID and reconcile its outcome before retrying. Do not treat generic model retries as an exactly-once guarantee.

Multiple workers

Per-session engine locks apply within one engine instance. If several workers can run the same session, route ownership or coordinate explicitly. Database persistence alone does not provide distributed execution serialization.

Limits and observability

Track model calls, tokens, errors, tool outcomes, latency and abandoned streams. Per-turn output limits do not cap total spend. Application budgets should account for retries and delegated agents.

Use deterministic test clients for execution contracts and real-provider tests for transport compatibility. Keep live-provider tests opt-in and avoid putting credentials in recorded fixtures.

Deployment readiness

Pin a release, validate your critical workflows on the built package, rehearse restart and recovery, and retain a rollback version. The documentation describes available interfaces; it does not certify every deployment configuration.

Troubleshooting

Capture a useful failure

AgentRunError carries the session ID and partial events. Keep the original cause where available. Record the installed version, provider, configured model and a minimal tool schema. Remove credentials and private user data before sharing a reproduction.

Close streams

If you stop reading early, explicitly close the async generator with contextlib.aclosing. Otherwise cleanup may wait for garbage collection. See streaming.

Inspect the actual public surface

Use API reference for the common interfaces and generated reference for exported symbols. Do not infer an implemented feature from a roadmap ticket or an old example.