FastAPI Tools Swarm Example

This page embeds the full source for examples/fastapi_tools_swarm.py.

"""Example: expose a code-defined swarm with shared runtime tools through FastAPI."""

from __future__ import annotations

import uvicorn

from swarmforge.api import create_swarm_app
from swarmforge.env import get_env, get_env_int
from swarmforge.swarm import SwarmDefinition, SwarmNode, SwarmVariable, function_tool


SHARED_STATE = {"tenant": "acme"}


async def lookup_order(order_id: str, context=None, state=None):
    """Fetch order status for a customer order."""

    return {
        "order_id": order_id,
        "tenant": (state or context.shared_state)["tenant"],
        "account_id": context.state.get("account_id"),
        "status": "shipped",
    }


TOOL_SWARM = SwarmDefinition(
    id="ops",
    name="Ops Swarm",
    nodes=[
        SwarmNode(
            id="ops-agent",
            node_key="ops",
            name="Ops",
            system_prompt="Use lookup_order when the user asks for an order update.",
            enabled_tools=[function_tool(handler=lookup_order)],
            is_entry_node=True,
        )
    ],
    variables=[
        SwarmVariable(
            key_name="account_id",
            description="Customer account identifier shared with tools",
            reducer_rule="overwrite",
        )
    ],
)


app = create_swarm_app(
    TOOL_SWARM,
    tool_state=SHARED_STATE,
    title="Swarm Tool API",
)


if __name__ == "__main__":
    uvicorn.run(
        app,
        host=get_env("SWARMFORGE_HOST", default="127.0.0.1") or "127.0.0.1",
        port=get_env_int("SWARMFORGE_PORT", default=8000),
    )

Run:

python examples/fastapi_tools_swarm.py

Communicate with the API using curl:

1. Create a session with an initial variable payload:

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

2. Send a message with a per-turn payload containing variables and user input:

curl -X POST http://127.0.0.1:8000/sessions/ops-1/messages \
    -H 'Content-Type: application/json' \
    -d '{
        "user_input": "Where is order 123?",
        "state": {
            "account_id": "ACME-991"
        }
    }'

Expected return (example shape):

{
    "events": [
        { "event": "open",       "data": { "currentAgent": "Ops" } },
        { "event": "tool_use",   "data": { "agentName": "Ops", "functionCalls": [{ "name": "lookup_order" }] } },
        { "event": "text_chunk", "data": { "text_chunk": "Order 123 ", "agentName": "Ops" } },
        { "event": "text_chunk", "data": { "text_chunk": "has shipped.", "agentName": "Ops" } },
        { "event": "done",       "data": { "fullText": "Order 123 has shipped.", "agentName": "Ops" } }
    ],
    "session": {
        "id": "ops-1"
    },
    "checkpoints": []
}

The exact event list and final text can vary by model, but the response always includes events, session, and checkpoints.

Fast mode variant

You can enable fast mode on the same swarm to skip the model turn and run lookup_order directly. Add behavior_config={"fast_mode_enabled": True} to the node, mark the tool with "fast_enabled": True, and optionally send per-turn arguments through fast_tool_params in the request state:

curl -X POST http://127.0.0.1:8000/sessions/ops-1/messages \
  -H 'Content-Type: application/json' \
  -d '{
    "user_input": "Track my order",
    "state": {
      "fast_tool_params": {
        "lookup_order": {"order_id": "A-99"}
      }
    }
  }'

When fast mode triggers, the event sequence becomes ["open", "fast_tool_use", "done"] instead of the usual model-turn events.