Agents SDK

Build agents with tools, conversations and streaming in Python.

Overview

The Agents SDK provides ask() for one-shot requests, Agent for conversations, and explicit engine components for application integration. Add tools, stream results, and choose in-memory, SQLite or PostgreSQL persistence.

Documentation · Quickstart · Tools · Production integration

It is the runtime layer in Fareground's family of open agent building blocks, alongside agent-id (identity), agent-messaging (encrypted agent-to-agent messaging), agent-memory (per-agent memory), and agent-knowledge (shared team knowledge).

What it deliberately does not do:

• Authentication — your app verifies users. You hand the router an authenticator that turns each request into a tenant + user, and the framework enforces it: a caller only ever reaches its own tenant's sessions and memory, and the tenant comes from the authenticator, never the request body. (Omit it and the API runs as a single shared tenant for local/dev — it warns loudly so you never ship that by accident.)
• The model — it talks to OpenAI, Anthropic, Google, and local models via Ollama. You bring the API key; pick any model at runtime, or let the framework detect a sensible default from your environment.
• The frontend — you build the chat UI; the framework streams events to it.

Package names. The distribution is published as fg-agents and the import package is fg_agents (e.g. from fg_agents import create_app). These are the names dependents rely on and are intentionally left unchanged; the repository name is agents-sdk.

Install

The core install is deliberately slim — engine, tools, ask/Agent, and in-memory persistence. Everything else is an extra:

pip install fg-agents                 # core (engine + tools + in-memory persistence)
pip install "fg-agents[web]"          # + FastAPI app/router (create_app)
pip install "fg-agents[postgres]"     # + PostgreSQL backend (SQLAlchemy + asyncpg)
pip install "fg-agents[sqlite]"       # + SQLite backend
pip install "fg-agents[openai]"       # + OpenAI (and OpenAI-compatible) providers
pip install "fg-agents[anthropic]"    # + Anthropic provider
pip install "fg-agents[google]"       # + Google Gemini provider
pip install "fg-agents[all]"          # everything (web + postgres + sqlite + all providers + scheduler)

Requires Python 3.11+.

Quickstart

With one API key env var set (export ANTHROPIC_API_KEY=...) and the matching provider extra installed (pip install "fg-agents[anthropic]"), this is the whole program:

import asyncio

from fg_agents import ask


async def main():
    print(await ask("What's 2+2?"))


asyncio.run(main())

Already inside async code — or in a REPL with top-level await (python -m asyncio, IPython, Jupyter)? Then it's just:

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

ask() detects the provider from your environment — ANTHROPIC_API_KEY, then OPENAI_API_KEY, then GOOGLE_API_KEY/GEMINI_API_KEY, then a local Ollama server — and picks a current model for it. Pass model="provider:model" to override. It takes tools, a system prompt, and a memory backend too, and stream() is its streaming twin:

from fg_agents import ask, stream

answer = await ask("Greet Ada.", tools=[greet], system_prompt="Be friendly.")

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

If you might break out of a stream() loop early, wrap it in contextlib.aclosing(...) so the ephemeral agent is closed deterministically.

Conversations: the Agent facade

ask() is one-shot. For multi-turn conversations, the Agent facade is one object that wires the LLM client, tool registry, persistence, and engine — usable as an async context manager. Tools are plain functions; the schema is derived from the signature and docstring:

def greet(name: str) -> str:
    """Greet someone by name."""
    return f"Hello, {name}!"

async with Agent(
    tools=[greet],                    # model= optional — auto-detected
    system_prompt="You are a friendly greeter.",
    memory="sqlite",   # persist sessions; "memory" (default) or "postgres:<url>"
) as agent:
    first = await agent.run("Please greet Ada.")
    second = await agent.run("Now greet Grace.")  # same conversation

run() returns an AgentRunResult (.text, .session_id, .events — str() is the answer text), and agent.stream() yields raw StreamEvents as they happen. Any failure in run() — an engine error event or a raised exception — surfaces as a single AgentRunError carrying the session_id and the partial events, with the original exception chained as __cause__. Each Agent instance keeps one conversation by default; pass session_id= per call to target another.

Going lower level

The facade is sugar over four objects you can wire yourself when you need full control (custom middleware, skills, shared registries across agents). Graduate gradually via agent.engine, agent.llm, agent.tools, and agent.repository — or build the stack directly:

import asyncio
from fg_agents import AgentDefinition, AgentEngine, AgentLLM, ToolRegistry, create_repository, tool


@tool(description="Greet someone by name")
def greet(name: str) -> str:
    return f"Hello, {name}!"


async def main():
    llm = AgentLLM()
    repo = create_repository("memory")   # or "sqlite", "postgres"
    await repo.initialize()             # explicit here; the facade does this lazily

    tools = ToolRegistry()
    tools.register_function(greet)

    agent = AgentDefinition(
        name="greeter",
        model="anthropic:claude-sonnet-4-6",  # required — no default model
        system_prompt="You are a friendly greeter. Use the greet tool when asked.",
        tools=["greet"],
    )

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


asyncio.run(main())

A full web backend

create_app returns a FastAPI application exposing routes to start sessions, post messages, and stream replies over SSE.

from fg_agents import AgentDefinition, ToolRegistry, create_app, tool


@tool(description="Search the knowledge base")
async def search(query: str) -> str:
    return f"Results for '{query}'..."


assistant = AgentDefinition(
    name="assistant",
    model="anthropic:claude-sonnet-4-6",
    system_prompt="You are a helpful assistant. Use your tools to answer.",
    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",  # production
)
# uvicorn myapp:app

The frontend then creates a session and streams messages:

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

See examples/ for five runnable apps, from a two-line facade hello-world to a complete chat UI (examples/04_web_app.py), plus frontend SSE clients in examples/frontend/.

Documentation

Start at docs/index.md for tutorials, concepts, provider setup, tools, hosting and troubleshooting.

• docs/api.md — reference for the public API surface
• docs/streaming.md — the SSE wire format and every event type (for frontend authors)
• docs/persistence.md — memory / SQLite / PostgreSQL backends and initialization semantics

Concepts / Building blocks

flowchart TB
    User([User in browser])
    subgraph Framework["Agent"]
        direction TB
        Skills["Skills<br/><sub>knowledge & playbooks</sub>"]
        Tools["Tools<br/><sub>actions the agent can take</sub>"]
        Subs["Sub-agents<br/><sub>specialists to delegate to</sub>"]
    end
    DB[("Your database<br/><sub>sessions, history</sub>")]
    LLM([Any model<br/>OpenAI · Anthropic · local])

    User <-->|SSE stream| Framework
    Framework <--> Skills
    Framework <--> Tools
    Framework <--> Subs
    Framework <--> LLM
    Framework <--> DB

Project structure

src/fg_agents/
    agent.py       Agent facade — one-object setup (quickstart tier)
    core/          Engine, LLM client, types, errors
    tools/         @tool decorator, registry, built-in tools, handlers
    orchestrator/  Multi-agent orchestration + sub-agent runner
    skills/        Skill loading and injection
    memory/        Working memory, context management, health scoring
    middleware/    Audit, token tracking, permissions, rate limit, loop guard
    persistence/   Multi-backend: in-memory, SQLite, PostgreSQL
    streaming/     SSE event definitions
    prompts/       System prompt construction
    scheduler/     Cron-scheduled agents
    api/           Service layer, FastAPI router, auth, schemas
    app.py         create_app() application factory
examples/          Five runnable apps + frontend SSE clients
docs/              Reference docs: API, streaming wire format, persistence
tests/             Test suite

Contributing

Development setup, tests, linting, and commit conventions live in CONTRIBUTING.md. See the changelog for what's changed.

_{Built by Fareground.
Licensed under Apache-2.0.}