FastAPI Swarm Example

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

"""Example: define a code-first swarm in Python and expose it through the fixed-swarm FastAPI API."""

from __future__ import annotations

from typing import Any, Dict

import uvicorn

from swarmforge.api import create_swarm_app
from swarmforge.env import get_env, get_env_int
from swarmforge.swarm import (
    SwarmDefinition,
    SwarmNode,
    SwarmNodeSubAgent,
    SwarmVariable,
)

SUPPORT_SWARM = SwarmDefinition(
    id="support",
    name="Support Swarm",
    nodes=[
        SwarmNode(
            id="triage",
            node_key="triage",
            name="Triage",
            system_prompt=(
                "You triage support requests. "
                "Transfer billing issues to Billing once the route is clear."
            ),
            sub_agents=[
                SwarmNodeSubAgent(
                    sub_agent="billing",
                    handoff_description="Transfer billing questions after confirming the request is billing-related.",
                    required_variables=["account_id"],
                )
            ],
            is_entry_node=True,
        ),
        SwarmNode(
            id="billing",
            node_key="billing",
            name="Billing",
            system_prompt="You handle billing issues clearly and directly.",
        ),
    ],
    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,
    extract_required_variables=extract_required_variables,
    title="Swarm 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_swarm.py