swarmforge 0.1.0

Headless framework for multi-agent chat runtime and evaluation.

SwarmForge

Headless Python framework for authoring, orchestrating, and evaluating multi-agent swarms.

The framework supports OpenAI-compatible hosted backends, with OpenRouter as the default and Google's Gemini OpenAI-compatible endpoint as a built-in alternative.

• Prompt-generation helpers for node, edge, and full-graph authoring
• Sample swarm-generation workflow based on prompt outputs plus validation/build helpers
• Graph validation and conversion from generated JSON into runtime models
• Store-backed chat orchestration with handoffs, queued intents, scoped state, live tools, and tool mocks
• Conversation tracing and deterministic evaluation utilities
• Heuristic scenario and intent generation for swarm evals
• Pure scoring helpers for routing, variables, tools, turn count, and agent coverage

Contents

• Motivation
• Install
• OpenRouter Setup
• Gemini OpenAI-Compat Setup
• Package Layout
• Core Concepts
• Authoring
• Chat Orchestration
• OpenRouter Client
• FastAPI API
• Backend Startup
• React Demo UI
• Gemini Client
• Evaluation
• Examples
• Development

Motivation

This project exists for teams that want a small, headless multi-agent runtime they can own end-to-end without committing to a larger platform or hiding orchestration logic behind framework internals.

The core idea is simple: define a swarm as explicit graph data, run it through a predictable orchestration loop, evaluate its behavior, and expose it through your own API surface. That keeps the system understandable in code review, easy to version in Git, and straightforward to embed inside an existing backend.

In practice, use this project when you want:

• a headless orchestration layer you can own end-to-end
• OpenRouter by default, with Gemini or other OpenAI-compatible backends behind the same interface
• prompt-generated or JSON-authored swarm definitions that can be versioned like application config
• explicit session-store injection instead of committing to one framework runtime
• lightweight FastAPI exposure for product APIs and internal tooling
• deterministic eval utilities for routing, handoffs, variables, and agent coverage

Install

cd swarmforge
pip install -e '.[dev]'
pytest

Python 3.11+ is required.

For the FastAPI server and demo UI, install the API extras too:

pip install -e '.[api]'

Documentation

This repository also ships with a VitePress docs site under docs/.

npm install
npm run docs:dev

Build the static docs:

npm run docs:build

Distribution

This package is published to PyPI as swarmforge.

One-time PyPI setup:

1. Create the swarmforge project on PyPI, or reserve the name if it does not exist yet.
2. In PyPI, configure Trusted Publishing for GitHub repository Rvey/swarm-forge.
3. Point that publisher at workflow .github/workflows/release.yml and environment pypi.
4. Confirm project.version in pyproject.toml matches the version you intend to release.

Local verification from the repository root:

pip install -e '.[dev]'
pytest
rm -rf dist/ build/
python3 -m build
python3 -m twine check dist/*

Artifacts are written to dist/.

Manual upload, if you want to publish directly from your machine instead of GitHub Actions:

rm -rf dist/ build/
python3 -m build
python3 -m twine check dist/*
python3 -m twine upload dist/*

Automatic publish via GitHub tag:

1. Bump project.version in pyproject.toml.
2. Commit and push the version change.
3. Create a tag that exactly matches the package version, with or without a leading v.
4. Push the tag.

Example:

python3 -m build
python3 -m twine check dist/*
git tag v0.1.0
git push origin main
git push origin v0.1.0

The release workflow builds the package, runs twine check, and publishes to PyPI through Trusted Publishing. It will fail if the pushed tag does not match project.version exactly after stripping an optional leading v.

Release checklist:

• run pytest
• remove old artifacts from dist/ and build/
• update project.version in pyproject.toml
• build with python3 -m build
• validate with python3 -m twine check dist/*
• create and push a matching release tag such as v0.1.0
• let the release workflow publish, or upload manually with python3 -m twine upload dist/*

OpenRouter Setup

The provider layer is OpenRouter-first and uses OpenRouter's OpenAI-compatible API.

Set:

export OPENROUTER_API_KEY=sk-or-...

Or place it in a local .env file in the repository root or current working directory:

OPENROUTER_API_KEY=sk-or-...

Optional attribution headers supported by OpenRouter:

export OPENROUTER_SITE_URL=https://your-app.example
export OPENROUTER_APP_NAME="Your App Name"

By default the framework uses:

• provider: openrouter
• model: openrouter/auto
• base URL: https://openrouter.ai/api/v1

Minimal client setup:

from swarmforge.evaluation.provider import ModelConfig, OpenAIClientWrapper

client = OpenAIClientWrapper(
    ModelConfig(
        provider="openrouter",
        model="openrouter/auto",
        site_url="https://your-app.example",
        app_name="Your App Name",
    )
)

For API server usage, the framework will read these values either from the current shell environment or from a local .env file before starting uvicorn.

Gemini OpenAI-Compat Setup

The same provider wrapper also supports Google's Gemini OpenAI-compatible endpoint documented here: OpenAI compatibility.

Set either:

export GEMINI_API_KEY=...

or:

export GOOGLE_API_KEY=...

You can also place either key in a local .env file:

GEMINI_API_KEY=...

Minimal Gemini setup:

from swarmforge.evaluation.provider import ModelConfig, OpenAIClientWrapper

client = OpenAIClientWrapper(
    ModelConfig(
        provider="gemini",
        model="gemini-3-flash-preview",
    )
)

This automatically uses Google's OpenAI-compatible base URL:

• https://generativelanguage.googleapis.com/v1beta/openai/

For API server usage, the framework will read either GEMINI_API_KEY or GOOGLE_API_KEY from the current shell environment or from a local .env file.

Package Layout

src/swarmforge/
  authoring/     Prompt templates + graph validators/builders
  evaluation/    Conversation runner + swarm eval helpers
  swarm/         Runtime models, stores, and orchestrator
examples/
  build_support_swarm.py
  fastapi_server.py
  openrouter_conversation.py
  run_support_swarm.py
  single_agent_swarm.py
  evaluate_support_swarm.py
demo-ui/
  React + Vite orchestration demo

Core Concepts

SwarmDefinition : The graph itself. Nodes, edges, variables, and a graph_version.

SwarmSession : Runtime state for a single user conversation. Tracks the current node, global/scoped variables, per-agent histories, and runtime context.

SessionStore : Persistence contract for sessions and checkpoints. The package ships with InMemorySessionStore.

process_swarm_stream(...) : Async orchestration loop for one user turn. It handles tool calls, handoffs, queued triage intents, scoped variable reducers, and checkpoints.

ModelConfig + OpenAIClientWrapper : Hosted model client configuration. Defaults to OpenRouter, supports Gemini via Google's OpenAI-compatible endpoint, and accepts optional provider-specific request defaults.

Authoring

Use the authoring helpers when an LLM generates node/edge/graph JSON and you want to validate it before execution.

This is also the project's swarm-generation path today: generation is provided as a sample prompt-driven workflow, then validated and converted into a runnable SwarmDefinition.

from swarmforge.authoring import build_swarm_definition

generated = {
    "nodes": [
        {
            "node_key": "triage",
            "name": "Triage",
            "intent": "Route support issues",
            "capabilities": ["Classify requests", "Route to specialists"],
            "persona": "",
            "is_entry_node": True,
        },
        {
            "node_key": "billing",
            "name": "Billing",
            "intent": "Handle billing questions",
            "capabilities": ["Explain invoices", "Resolve payment issues"],
            "persona": "Direct and calm",
            "is_entry_node": False,
        },
    ],
    "edges": [
        {
            "source_node_key": "triage",
            "target_node_key": "billing",
            "handoff_description": "Transfer only after confirming the request is billing-related.",
            "required_variables": ["account_id"],
        }
    ],
}

swarm = build_swarm_definition(generated, swarm_id="support", name="Support Swarm")

The module also exports reusable meta prompts:

• META_PROMPT_NODE
• META_PROMPT_EDGE
• META_PROMPT_SWARM

These are transport-agnostic templates. Use them with OpenRouter or any other model client you prefer.

Chat Orchestration

The orchestrator is intentionally callback-based. You provide the model-driving function for a single agent iteration; the framework handles the swarm runtime.

If you only need one assistant, you can still use the swarm runtime with a single entry node and no edges. That keeps the same session/checkpoint model without introducing handoffs.

import asyncio

from swarmforge.swarm import (
    InMemorySessionStore,
    SwarmDefinition,
    SwarmEdge,
    SwarmNode,
    SwarmSession,
    process_swarm_stream,
)


swarm = SwarmDefinition(
    id="support",
    name="Support Swarm",
    nodes=[
        SwarmNode(
            id="triage",
            node_key="triage",
            name="Triage",
            intent="Route support issues",
            system_prompt="You are the triage agent.",
            is_entry_node=True,
        ),
        SwarmNode(
            id="billing",
            node_key="billing",
            name="Billing",
            intent="Handle billing questions",
            system_prompt="You handle billing.",
        ),
    ],
    edges=[
        SwarmEdge(
            id="triage->billing",
            source_node_id="triage",
            target_node_id="billing",
            handoff_description="Transfer billing issues after intent confirmation.",
            required_variables=["account_id"],
        )
    ],
)

session = SwarmSession(id="session-1", swarm=swarm)
store = InMemorySessionStore()


async def stream_agent_iteration(*, agent_node, contents, config):
    if agent_node.node_key == "triage":
        yield "", [{"id": "handoff-1", "name": "transfer_to_agent", "args": {"target_node_key": "billing"}}], None
        return
    yield "I can help with that billing issue.", [], None


async def extract_required_variables(**_kwargs):
    return {"account_id": "ACME-991"}


async def main():
    async for event in process_swarm_stream(
        session,
        "I need help with a charge on my account.",
        store=store,
        stream_agent_iteration=stream_agent_iteration,
        extract_required_variables=extract_required_variables,
    ):
        print(event)


asyncio.run(main())

Runtime behavior includes:

• per-agent histories and context contracts
• automatic transfer_to_agent tool injection from graph edges
• reducer-based variable application
• explicit tool_mock_required events when a tool is called without a configured mock
• queued-intent routing for triage agents that emit structured multi-intent JSON
• checkpoints after handoffs, blocked handoffs, tool-mock stops, and agent responses

Live Tool Handlers

For Python integrations, tools can run real code instead of mock responses. You can attach a handler directly to a tool definition or register tools once at the framework level and reuse them across agents. JSON schema is inferred automatically from the Python function signature, and descriptions are pulled from the function docstring so you do not need to hand-write parameters for common cases.

The runtime injects non-model arguments when the handler asks for them. In addition to model-provided params, a tool can accept context, state, shared_state, session, agent_node, user_input, or visible_global_variables without exposing those fields in the tool schema.

import asyncio

from swarmforge.swarm import (
    FrameworkToolRegistry,
    InMemorySessionStore,
    SwarmDefinition,
    SwarmNode,
    SwarmSession,
    process_swarm_stream,
)


tools = FrameworkToolRegistry(shared_state={"tenant": "acme"})


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

    Args:
        order_id: The order identifier.
    """

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


swarm = SwarmDefinition(
    id="ops",
    name="Ops Swarm",
    nodes=[
        SwarmNode(
            id="ops-agent",
            node_key="ops",
            name="Ops",
            intent="Handle order lookups",
            system_prompt="Use tools when an order lookup is needed.",
            enabled_tools=tools.tools("lookup_order"),
            is_entry_node=True,
        )
    ],
)


async def stream_agent_iteration(*, agent_node, contents, config):
    del agent_node, contents, config
    yield "", [{"id": "tool-1", "name": "lookup_order", "args": {"order_id": "123"}}], None
    yield "Order 123 has shipped.", [], None


async def main():
    session = SwarmSession(id="session-1", swarm=swarm, global_variables={"account_id": "ACME-991"})
    store = InMemorySessionStore()
    async for event in process_swarm_stream(
        session,
        "Where is order 123?",
        store=store,
        stream_agent_iteration=stream_agent_iteration,
        tool_registry=tools,
    ):
        print(event)


asyncio.run(main())

If you prefer not to use the registry helper, function_tool(handler=my_func) also infers the schema directly from the callable.

For runtime-only dependencies such as HTTP clients, auth tokens, or service objects, pass tool_state=... into process_swarm_stream(...) or create_fastapi_app(...). That state is available to handlers through the injected state or shared_state parameter and stays outside the agent-visible session state.

OpenRouter Client

For direct LLM-backed conversation and simulation flows, use the provider wrapper with the conversation runner:

from swarmforge.evaluation.provider import ModelConfig, OpenAIClientWrapper
from swarmforge.evaluation.runner.conversation import ConversationRunner

client = OpenAIClientWrapper(
    ModelConfig(
        provider="openrouter",
        model="openrouter/auto",
        site_url="https://your-app.example",
        app_name="Your App Name",
    )
)
runner = ConversationRunner(
    client=client,
    system_prompt="You are a concise assistant.",
)
trace = runner.run_conversation(
    suite_id="demo",
    case_id="single-turn",
    conversation=[{"user_input": "Explain why OpenRouter is useful for agent workloads."}],
)
print(trace.turns[-1].assistant_response)

This uses OpenRouter's OpenAI-compatible /api/v1/chat/completions API and sends optional HTTP-Referer and X-OpenRouter-Title headers when configured.

FastAPI API

The framework can also be exposed as an HTTP API with FastAPI.

Install the API extras:

pip install -e '.[api]'

Backend Startup

If you installed the package in editable mode, run the API server from the repository root with:

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

If you did not install the package and want to run directly from source, use:

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

Equivalent form:

PYTHONPATH=src uvicorn swarmforge.api.fastapi:create_fastapi_app --factory --reload

Typical backend shell setup for OpenRouter:

cd swarmforge
pip install -e '.[api]'
export OPENROUTER_API_KEY=sk-or-...
export OPENROUTER_SITE_URL=http://127.0.0.1:5174
export OPENROUTER_APP_NAME="SwarmForge Demo"
uvicorn swarmforge.api.fastapi:create_fastapi_app --factory --reload

Equivalent .env setup:

OPENROUTER_API_KEY=sk-or-...
OPENROUTER_SITE_URL=http://127.0.0.1:5174
OPENROUTER_APP_NAME="SwarmForge Demo"

Typical backend shell setup for Gemini OpenAI-compat:

cd swarmforge
pip install -e '.[api]'
export GEMINI_API_KEY=...
uvicorn swarmforge.api.fastapi:create_fastapi_app --factory --reload

Equivalent .env setup:

GEMINI_API_KEY=...

Create an app:

from swarmforge.api import create_fastapi_app

app = create_fastapi_app()

By default the FastAPI integration uses in-memory storage.

If you want persistence, inject your own SessionStore backed by a database:

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())

Run it with:

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

Available endpoints:

• GET /health
• POST /v1/swarm/run Runs one user turn against a temporary session created from the provided swarm definition.
• POST /v1/swarm/run/stream Runs one user turn and streams incremental swarm events as Server-Sent Events.
• POST /v1/sessions Creates a session for a swarm definition.
• GET /v1/sessions/{session_id} Returns session state and checkpoints.
• POST /v1/sessions/{session_id}/messages Sends a user message through an existing session.
• POST /v1/sessions/{session_id}/messages/stream Sends a user message through an existing session and streams incremental swarm events as Server-Sent Events.

Stateless example:

curl -X POST http://127.0.0.1:8000/v1/swarm/run \
  -H 'Content-Type: application/json' \
  -d '{
    "swarm": {
      "id": "single-agent",
      "name": "Single Agent",
      "nodes": [
        {
          "node_key": "assistant",
          "name": "Assistant",
          "system_prompt": "You are a helpful assistant.",
          "intent": "Handle the full request",
          "capabilities": ["Answer questions"],
          "is_entry_node": true
        }
      ],
      "edges": [],
      "variables": []
    },
    "user_input": "Give me a concise answer."
  }'

The server uses the same provider layer as the library runtime. You can keep OpenRouter as default or pass a provider override in the request body to use Gemini instead.

SSE responses emit the native swarm runtime events such as open, tool_use, handoff, chunk, and done, followed by session and checkpoints events for the final persisted state.

The generated OpenAPI schema includes ready-to-try request examples for both OpenRouter and Gemini on the swarm run and session message endpoints.

React Demo UI

A minimal React + Vite demo is included in demo-ui/. It talks directly to the FastAPI orchestration layer and supports:

• session or stateless mode
• OpenRouter or Gemini provider selection
• editable swarm JSON
• normal JSON responses or SSE event streaming
• live session, event, and checkpoint inspection

Run the backend:

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

Run the demo UI:

cd demo-ui
npm install
npm run dev

Then open the Vite URL, keep the API base set to http://127.0.0.1:8000, and use the control panel to create a session or run the swarm directly.

Gemini Client

Gemini works through the same wrapper by changing only the provider and model:

from swarmforge.evaluation.provider import ModelConfig, OpenAIClientWrapper
from swarmforge.evaluation.runner.conversation import ConversationRunner

client = OpenAIClientWrapper(
    ModelConfig(
        provider="gemini",
        model="gemini-3-flash-preview",
        default_chat_params={"reasoning_effort": "low"},
    )
)
runner = ConversationRunner(
    client=client,
    system_prompt="You are a concise assistant.",
)
trace = runner.run_conversation(
    suite_id="demo",
    case_id="gemini-single-turn",
    conversation=[{"user_input": "Explain how tool calling works in an OpenAI-compatible API."}],
)
print(trace.turns[-1].assistant_response)

If you need Gemini-specific OpenAI-compatible fields such as extra_body, pass them through the client:

response = client.chat_completion(
    messages=[{"role": "user", "content": "Explain to me how AI works"}],
    extra_params={
        "extra_body": {
            "google": {
                "thinking_config": {
                    "include_thoughts": True,
                }
            }
        }
    },
)

Evaluation

The swarm eval module is pure Python. It does not depend on any web framework, billing layer, or product service.

from swarmforge.evaluation import (
    build_graph_snapshot,
    build_heuristic_swarm_intents,
    build_intent_based_swarm_scenario_seeds,
    classify_scenario_feasibility,
)

graph_snapshot = build_graph_snapshot(swarm)
intents = build_heuristic_swarm_intents(graph_snapshot, num_intents=3)
scenario_seeds = build_intent_based_swarm_scenario_seeds(
    graph_snapshot,
    [intent["title"] for intent in intents],
    num_scenarios=3,
    min_turns=2,
)

for seed in scenario_seeds:
    print(seed["starting_prompt"], classify_scenario_feasibility(graph_snapshot, seed))

The evaluation helpers cover:

• graph snapshots and routing-path analysis
• heuristic intent generation
• heuristic scenario-seed generation
• normalization of AI-produced seeds
• simulator-turn context shaping
• artifact scoring with evaluate_scenario_artifacts(...)

Examples

See the runnable scripts in examples/build_support_swarm.py, examples/fastapi_server.py, examples/openrouter_conversation.py, examples/run_support_swarm.py, examples/single_agent_swarm.py, and examples/evaluate_support_swarm.py.

Development

Run the full test suite:

pytest -q

The framework currently focuses on reusable orchestration and evaluation mechanics. Product-specific app adapters remain outside this repository by design.