FastAPI

This guide explains how to expose SwarmForge through FastAPI, choose between the generic and bound app shapes, and manage state, stores, and tool execution over HTTP. It does not explain swarm authoring from scratch or provider selection in depth.

This guide is for backend developers who want to add an HTTP layer on top of an existing swarm runtime. It assumes that you already understand the basic runtime objects and can configure a provider-backed model on the server.

After reading this guide, you should be able to:

choose between create_fastapi_app(...) and create_swarm_app(...)
configure server-side model defaults
manage session state and persistence
add tools to single-agent and multi-agent FastAPI apps

The swarmforge.api package exposes the runtime through FastAPI without changing the underlying orchestration model. The HTTP layer still uses the same session, handoff, tool, checkpoint, and event flow that process_swarm_stream(...) uses in direct Python integrations.

FastAPI overview

Install

bash

pip install "swarmforge[api]"

Before sending requests that should hit a real hosted model, configure the provider on the server side:

MODEL_PROVIDER
LLM_MODEL
the matching provider API key such as OPENROUTER_API_KEY, GEMINI_API_KEY, GOOGLE_API_KEY, or OPENAI_API_KEY

Optional OpenRouter attribution:

OPENROUTER_SITE_URL
OPENROUTER_APP_NAME

Optional bind config used by the example servers:

SWARMFORGE_HOST default: 127.0.0.1
SWARMFORGE_PORT default: 8000

Start the server

Installed-package path:

bash

uvicorn swarmforge.api.fastapi:create_fastapi_app --factory --reload

Repository-source path:

bash

uvicorn --app-dir src swarmforge.api.fastapi:create_fastapi_app --factory --reload

Useful companion examples:

examples/fastapi_server.py for the generic JSON transport
examples/fastapi_swarm.py for the code-defined bound-swarm pattern

API shapes

SwarmForge supports two FastAPI integration shapes:

Factory	Who defines the swarm	Route style	Best fit
`create_fastapi_app(...)`	The client sends swarm JSON	`/v1/...`	graph builders, visual editors, dynamic or multi-tenant swarms
`create_swarm_app(...)`	Your Python app defines the swarm once	`/sessions/...` and `/swarm`	product backends with a fixed single-agent or multi-agent workflow

create_fastapi_app(...) accepts:

default_model_config
session_store
turn_runner_factory
extract_required_variables
state_manager_factory
tool_registry
tool_state
cors_allow_origins

global_variable_manager_factory still works as a compatibility alias.

Both FastAPI factories use the server-side default_model_config when provided, or ModelConfig() from environment variables when omitted. The HTTP request payload does not select or override the provider.

Use create_fastapi_app(...) when the swarm itself is request data. Use create_swarm_app(...) when your backend owns the swarm definition and wants a smaller, typed route surface.

FastAPI stores

All session-backed FastAPI endpoints persist two runtime artifacts:

the current SwarmSession
the ordered list of SessionCheckpoint records produced during execution

InMemorySessionStore

create_fastapi_app(...) and create_swarm_app(...) both default to InMemorySessionStore().

That is the right default for:

local development
tests
demos
single-process deployments

Behavior:

sessions are keyed in memory by session id
checkpoints are appended in memory and returned in order
all state is lost on process restart
state is not shared across multiple workers or containers

Example:

python

from swarmforge.api import create_fastapi_app
from swarmforge.swarm import InMemorySessionStore


app = create_fastapi_app(session_store=InMemorySessionStore())

SessionStore contract

If you need persistence across restarts or multiple API instances, provide your own SessionStore.

Required operations:

get_session(session_id)
save_session(session)
append_checkpoint(checkpoint)
list_checkpoints(session_id)

Example skeleton:

python

from swarmforge.api import create_fastapi_app
from swarmforge.swarm import SessionStore


class YourDbSessionStore(SessionStore):
    async def get_session(self, session_id: str):
        ...

    async def save_session(self, session):
        ...

    async def append_checkpoint(self, checkpoint):
        ...

    async def list_checkpoints(self, session_id: str):
        ...


app = create_fastapi_app(session_store=YourDbSessionStore())

Use a custom store when you need:

durable sessions after restarts
shared sessions across multiple API instances
checkpoint audit trails
database-backed persistence with Postgres, Redis, DynamoDB, or another system

FastAPI state model

Client-supplied request state is the normal way to pass application facts such as account_id, priority, tenant_id, or region into the runtime.

Request state payloads

Both FastAPI factories accept state updates in request bodies:

json

{
  "state": {
    "account_id": "ACME-991",
    "priority": "high"
  },
  "state_mode": "merge"
}

Use:

state to send values
state_mode with merge or replace

For backward compatibility, the API still accepts variables, variable_mode, global_variables, and global_variable_mode, but new clients should use state and state_mode.

Those values become visible in:

tool handlers through context.state or context.visible_state
dynamic prompts through SystemPromptContext.state
the active turn config through config.state

Provider selection is not part of the request state or message body. FastAPI uses the provider configured on the server through default_model_config or the standard provider environment variables.

State and runtime endpoints

Both FastAPI factories expose state inspection and update routes:

GET /.../sessions/{session_id}/state
PATCH /.../sessions/{session_id}/state

Legacy aliases remain available:

GET /.../sessions/{session_id}/variables
PATCH /.../sessions/{session_id}/variables

Runtime inspection is separate:

GET /.../sessions/{session_id}/runtime

The runtime payload returns:

reducer-aware global state
per-node history
per-node scratchpad/context
direct entry_node and current_node runtime views

This is runtime-visible working state, not opaque model chain-of-thought.

Optional state validation

SessionStateManager is the preferred name for the reducer-aware runtime state contract. GlobalVariableManager remains available as a compatibility alias.

Use a custom manager when you need to:

validate values before writes
normalize or coerce values before reducers run
add custom reducer behavior
keep HTTP and in-process state rules on the same contract

Example:

python

from swarmforge.api import create_swarm_app
from swarmforge.swarm import SessionStateManager


class ValidatingState(SessionStateManager):
    def normalize_value(self, key, value, *, session, current_node=None):
        if key == "account_id":
            return str(value).strip().upper()
        return value

    def validate_value(self, key, value, *, session, current_node=None, reducer_rule):
        if key == "priority" and value not in {"low", "normal", "high"}:
            raise ValueError("priority must be low, normal, or high")


app = create_swarm_app(
    SUPPORT_SWARM,
    state_manager=ValidatingState.from_swarm(SUPPORT_SWARM),
)

Single-agent FastAPI

Use a single-agent FastAPI app when your backend owns one assistant workflow and you want typed routes without sending swarm JSON on every request.

Single-agent usage

The usual shape is:

define one SwarmNode
bind it once with create_swarm_app(...)
optionally attach Python tools
create sessions and send messages through /sessions/...

This is a good fit for:

internal copilots
account assistants
single-lane support bots
APIs that should not accept arbitrary swarm definitions from clients

Single-agent example

python

from swarmforge.api import create_swarm_app
from swarmforge.evaluation.provider import ModelConfig
from swarmforge.swarm import SwarmDefinition, SwarmNode


ASSISTANT_SWARM = SwarmDefinition(
    id="assistant",
    name="Assistant Swarm",
    nodes=[
        SwarmNode(
            id="assistant",
            node_key="assistant",
            name="Assistant",
            system_prompt="You are a concise assistant.",
            is_entry_node=True,
        )
    ],
)


app = create_swarm_app(
    ASSISTANT_SWARM,
    default_model_config=ModelConfig(),
    title="Single-Agent API",
)

That app exposes:

GET /swarm
POST /sessions
GET /sessions/{session_id}
POST /sessions/{session_id}/messages
POST /sessions/{session_id}/messages/stream
the matching state and runtime endpoints

Single-agent tool usage

For a single-agent FastAPI app, the simplest path is:

declare the tool on the node with function_tool(...)
pass the callable through tool_registry when you create the app
let the model call the tool during a normal /sessions/{session_id}/messages request

python

from swarmforge.api import create_swarm_app
from swarmforge.evaluation.provider import ModelConfig
from swarmforge.swarm import SwarmDefinition, SwarmNode, function_tool


async def lookup_order(order_id: str, context=None, state=None):
    return {
        "order_id": order_id,
        "account_id": state.get("account_id"),
        "status": "shipped",
    }


ASSISTANT_SWARM = SwarmDefinition(
    id="assistant",
    name="Assistant Swarm",
    nodes=[
        SwarmNode(
            id="assistant",
            node_key="assistant",
            name="Assistant",
            system_prompt="Always call lookup_order before answering order-status questions.",
            enabled_tools=[function_tool(handler=lookup_order)],
            is_entry_node=True,
        )
    ],
)


app = create_swarm_app(
    ASSISTANT_SWARM,
    default_model_config=ModelConfig(),
    tool_registry={"lookup_order": lookup_order},
    title="Single-Agent API",
)

The request payload does not change for tools. The user still sends a normal message, and the runtime emits tool-related events if the model decides to call the tool.

Single-agent payloads

Create a session:

bash

curl -X POST http://127.0.0.1:8000/sessions \
  -H 'Content-Type: application/json' \
  -d '{
    "session_id": "assistant-1",
    "state": {
      "account_id": "ACME-991"
    }
  }'

Send a message:

bash

curl -X POST http://127.0.0.1:8000/sessions/assistant-1/messages \
  -H 'Content-Type: application/json' \
  -d '{
    "user_input": "Give me a concise account summary.",
    "state": {
      "priority": "high"
    },
    "state_mode": "merge"
  }'

Typical request body fields for the bound single-agent API:

Route	Required fields	Optional fields
`POST /sessions`	none	`session_id`, `state`
`POST /sessions/{session_id}/messages`	`user_input`	`state`, `state_mode`

Multi-agent FastAPI

Use a multi-agent FastAPI app when your backend owns routing logic and wants stable HTTP routes while the swarm decides how to hand off between specialized nodes.

Multi-agent usage

The usual shape is:

define multiple SwarmNode entries plus SwarmEdge handoffs
add required_variables where routing depends on known facts
optionally attach Python tools to one or more nodes
optionally provide extract_required_variables(...)
bind the swarm once with create_swarm_app(...)

This is a good fit for:

support flows with triage and specialists
billing, refund, fraud, or escalation routing
product backends that need inspectable handoff events

Multi-agent example

python

from typing import Any, Dict

from swarmforge.api import create_swarm_app
from swarmforge.evaluation.provider import ModelConfig
from swarmforge.swarm import SwarmDefinition, SwarmEdge, SwarmNode, SwarmVariable


SUPPORT_SWARM = SwarmDefinition(
    id="support",
    name="Support Swarm",
    nodes=[
        SwarmNode(
            id="triage",
            node_key="triage",
            name="Triage",
            system_prompt="You triage support requests and transfer when the right specialist is clear.",
            is_entry_node=True,
        ),
        SwarmNode(
            id="billing",
            node_key="billing",
            name="Billing",
            system_prompt="You handle billing issues clearly and directly.",
        ),
    ],
    edges=[
        SwarmEdge(
            id="triage->billing",
            source_node_id="triage",
            target_node_id="billing",
            handoff_description="Transfer billing issues after confirmation.",
            required_variables=["account_id"],
        )
    ],
    variables=[
        SwarmVariable(
            key_name="account_id",
            description="Customer account identifier",
            reducer_rule="overwrite",
        )
    ],
)


async def extract_required_variables(user_input: str = "", **_kwargs) -> Dict[str, Any]:
    if "ACME-991" in user_input:
        return {"account_id": "ACME-991"}
    return {}


app = create_swarm_app(
    SUPPORT_SWARM,
    default_model_config=ModelConfig(),
    extract_required_variables=extract_required_variables,
    title="Support API",
)

That app exposes the same session routes as the single-agent bound app plus:

handoff-capable runtime behavior
GET /swarm to inspect the bound swarm definition
state and runtime endpoints that show the active node and accumulated state

Multi-agent tool usage

Multi-agent FastAPI apps can use both:

your own Python tools through enabled_tools plus tool_registry
the runtime-injected transfer_to_agent tool that comes from graph edges

python

from typing import Any, Dict

from swarmforge.api import create_swarm_app
from swarmforge.evaluation.provider import ModelConfig
from swarmforge.swarm import SwarmDefinition, SwarmEdge, SwarmNode, SwarmVariable, function_tool


async def lookup_invoice(invoice_id: str, context=None, state=None):
    return {
        "invoice_id": invoice_id,
        "account_id": state.get("account_id"),
        "status": "overdue",
    }


SUPPORT_SWARM = SwarmDefinition(
    id="support",
    name="Support Swarm",
    nodes=[
        SwarmNode(
            id="triage",
            node_key="triage",
            name="Triage",
            system_prompt=(
                "Use lookup_invoice when invoice details are needed. "
                "Call transfer_to_agent when the request should move to Billing."
            ),
            enabled_tools=[function_tool(handler=lookup_invoice)],
            is_entry_node=True,
        ),
        SwarmNode(
            id="billing",
            node_key="billing",
            name="Billing",
            system_prompt="You handle billing issues clearly and directly.",
        ),
    ],
    edges=[
        SwarmEdge(
            id="triage->billing",
            source_node_id="triage",
            target_node_id="billing",
            handoff_description="Transfer billing issues after confirmation.",
            required_variables=["account_id"],
        )
    ],
    variables=[SwarmVariable(key_name="account_id", reducer_rule="overwrite")],
)


async def extract_required_variables(user_input: str = "", **_kwargs) -> Dict[str, Any]:
    if "ACME-991" in user_input:
        return {"account_id": "ACME-991"}
    return {}


app = create_swarm_app(
    SUPPORT_SWARM,
    default_model_config=ModelConfig(),
    extract_required_variables=extract_required_variables,
    tool_registry={"lookup_invoice": lookup_invoice},
    title="Support API",
)

As with the single-agent case, clients still send normal message payloads. Tool execution and handoffs happen inside the runtime and show up in the event stream and runtime/session responses.

Multi-agent payloads

Create a session:

bash

curl -X POST http://127.0.0.1:8000/sessions \
  -H 'Content-Type: application/json' \
  -d '{
    "session_id": "support-session-1",
    "state": {
      "priority": "high"
    }
  }'

Send a message:

bash

curl -X POST http://127.0.0.1:8000/sessions/support-session-1/messages \
  -H 'Content-Type: application/json' \
  -d '{
    "user_input": "I need help with a charge on account ACME-991."
  }'

Inspect runtime state after the handoff:

bash

curl http://127.0.0.1:8000/sessions/support-session-1/runtime

Typical request body fields for the bound multi-agent API:

Route	Required fields	Optional fields
`POST /sessions`	none	`session_id`, `state`
`POST /sessions/{session_id}/messages`	`user_input`	`state`, `state_mode`

Endpoint surface

Generic JSON transport endpoints

Routes exposed by create_fastapi_app(...):

GET /health
POST /v1/swarm/run
POST /v1/swarm/run/stream
POST /v1/sessions
GET /v1/sessions/{session_id}
GET /v1/sessions/{session_id}/state
PATCH /v1/sessions/{session_id}/state
GET /v1/sessions/{session_id}/variables
PATCH /v1/sessions/{session_id}/variables
GET /v1/sessions/{session_id}/runtime
POST /v1/sessions/{session_id}/messages
POST /v1/sessions/{session_id}/messages/stream

Use this shape when the swarm is dynamic request data.

Bound swarm endpoints

Routes exposed by create_swarm_app(...):

GET /health
GET /swarm
POST /sessions
GET /sessions/{session_id}
GET /sessions/{session_id}/state
PATCH /sessions/{session_id}/state
GET /sessions/{session_id}/variables
PATCH /sessions/{session_id}/variables
GET /sessions/{session_id}/runtime
POST /sessions/{session_id}/messages
POST /sessions/{session_id}/messages/stream

Use this shape when the swarm is code-defined once inside your FastAPI service.

Transport behavior

Both FastAPI shapes support:

SSE streaming for incremental runtime events
CORS support for browser clients
OpenAPI request examples that assume server-configured provider defaults
injected session persistence through the SessionStore interface

Streaming endpoints emit runtime events first, then final session and checkpoints events.

API

Single-Agent FastAPI

Multi-Agent FastAPI

Evaluation

SDK Evaluation Tutorial

FastAPI Evaluation Tutorial

Examples and Artifacts

FastAPI

FastAPI overview

Install

Start the server

API shapes

FastAPI stores

InMemorySessionStore

SessionStore contract

FastAPI state model

Request state payloads

State and runtime endpoints

Optional state validation

Single-agent FastAPI

Single-agent usage

Single-agent example

Single-agent tool usage

Single-agent payloads

Multi-agent FastAPI

Multi-agent usage

Multi-agent example

Multi-agent tool usage

Multi-agent payloads

Endpoint surface

Generic JSON transport endpoints

Bound swarm endpoints

Transport behavior

Single-Agent FastAPI

Multi-Agent FastAPI

SDK Evaluation Tutorial

FastAPI Evaluation Tutorial

Examples and Artifacts

FastAPI ​

FastAPI overview ​

Install ​

Start the server ​

API shapes ​

FastAPI stores ​

InMemorySessionStore ​

SessionStore contract ​

FastAPI state model ​

Request state payloads ​

State and runtime endpoints ​

Optional state validation ​

Single-agent FastAPI ​

Single-agent usage ​

Single-agent example ​

Single-agent tool usage ​

Single-agent payloads ​

Multi-agent FastAPI ​

Multi-agent usage ​

Multi-agent example ​

Multi-agent tool usage ​

Multi-agent payloads ​

Endpoint surface ​

Generic JSON transport endpoints ​

Bound swarm endpoints ​

Transport behavior ​

FastAPI

FastAPI overview

Install

Start the server

API shapes

FastAPI stores

InMemorySessionStore

SessionStore contract

FastAPI state model

Request state payloads

State and runtime endpoints

Optional state validation

Single-agent FastAPI

Single-agent usage

Single-agent example

Single-agent tool usage

Single-agent payloads

Multi-agent FastAPI

Multi-agent usage

Multi-agent example

Multi-agent tool usage

Multi-agent payloads

Endpoint surface

Generic JSON transport endpoints

Bound swarm endpoints

Transport behavior