swarmforge 0.7.0

Headless framework for multi-agent chat runtime and evaluation.

SwarmForge

SwarmForge is a Python package for authoring, running, and evaluating multi-agent swarms. You define the swarm graph, provide the model-turn callback, and keep sessions, handoffs, tools, checkpoints, and evaluation artifacts under your control.

• Explicit graph-based multi-agent runtime
• Python tool execution with inferred JSON schema
• OpenRouter, Gemini, and other OpenAI-compatible provider support
• FastAPI transport for stateless and session-backed HTTP flows
• Evaluation helpers for graph snapshots, scenario seeds, and artifact scoring

Install

Python 3.11+ is required.

pip install swarmforge

If you want the FastAPI transport too:

pip install "swarmforge[api]"

Provider-backed examples, the demo UI, and local API runs load a nearby .env automatically. Copy .env.example to .env, then set MODEL_PROVIDER, LLM_MODEL, and the matching API key before using any runnable example:

cp .env.example .env

Quick Start

The shortest path is a single-node swarm with a real provider-backed turn runner:

import asyncio
import json

from swarmforge.env import require_env_vars
from swarmforge.evaluation.provider import ModelConfig, OpenAIClientWrapper
from swarmforge.swarm import (
    AgentTurnConfig,
    AgentTurnResult,
    InMemorySessionStore,
    SwarmDefinition,
    SwarmNode,
    SwarmSession,
    process_swarm_stream,
)


class ProviderBackedTurnRunner:
    def __init__(self) -> None:
        self.client = OpenAIClientWrapper(ModelConfig())

    async def run_turn(self, *, agent_node, contents, config: AgentTurnConfig):
        del agent_node
        messages = []
        if config.system_instruction:
            messages.append({"role": "system", "content": config.system_instruction})

        for item in contents:
            role = str(item.get("role") or "user").strip()
            if role == "model":
                role = "assistant"
            message = {"role": role, "content": item.get("content", "")}
            if role == "tool":
                if item.get("tool_call_id"):
                    message["tool_call_id"] = item.get("tool_call_id")
                if item.get("name"):
                    message["name"] = item.get("name")
            messages.append(message)

        response = self.client.chat_completion(messages=messages, tools=config.tools)
        assistant_message = response.choices[0].message
        tool_calls = []
        if assistant_message.tool_calls:
            for tool_call in assistant_message.tool_calls:
                args = tool_call.function.arguments
                if not isinstance(args, dict):
                    args = json.loads(args or "{}")
                tool_calls.append(
                    {
                        "id": tool_call.id,
                        "name": tool_call.function.name,
                        "args": args,
                    }
                )

        return AgentTurnResult(
            response_text=assistant_message.content or "",
            tool_calls=tool_calls,
            raw_response=response,
        )


swarm = SwarmDefinition(
    id="assistant",
    name="Assistant Swarm",
    nodes=[
        SwarmNode(
            id="assistant",
            node_key="assistant",
            name="Assistant",
            intent="Handle general requests",
            system_prompt="You are a concise assistant.",
            capabilities=["Answer questions"],
            is_entry_node=True,
        )
    ],
)


async def main():
    require_env_vars("MODEL_PROVIDER", "LLM_MODEL")
    session = SwarmSession(id="session-1", swarm=swarm)
    store = InMemorySessionStore()
    async for event in process_swarm_stream(
        session,
        "Give me a concise summary.",
        store=store,
        turn_runner=ProviderBackedTurnRunner(),
    ):
        print(json.dumps(event, indent=2))


if __name__ == "__main__":
    asyncio.run(main())

The final done event contains the real model output, so the wording varies by provider and model. When you are ready to add routing, continue with the multi-agent flow in the docs.

Package Surfaces

• swarmforge.swarm Runtime models, session state, orchestration, tool execution, and stores.
• swarmforge.authoring Prompt templates, payload validation, and graph compilation helpers.
• swarmforge.evaluation Graph snapshots, scenario generation, feasibility checks, and artifact scoring.
• swarmforge.api FastAPI application factory built on the same runtime primitives.

Providers

SwarmForge ships with an OpenAI-compatible provider wrapper. OpenRouter is the default path, and Gemini is built in as an alternative mode.

Start from the repository .env.example and explicitly set both the provider and the model you want to use.

OpenRouter .env:

MODEL_PROVIDER=openrouter
LLM_MODEL=openrouter/auto
OPENROUTER_API_KEY=sk-or-...
OPENROUTER_SITE_URL=https://your-app.example
OPENROUTER_APP_NAME="Your App Name"

Gemini .env:

MODEL_PROVIDER=gemini
LLM_MODEL=gemini-3-flash-preview
GEMINI_API_KEY=...

Minimal client setup:

from swarmforge.evaluation.provider import ModelConfig, OpenAIClientWrapper

client = OpenAIClientWrapper(ModelConfig())

ModelConfig() reads MODEL_PROVIDER, LLM_MODEL, and the matching API key from .env or the shell environment.

FastAPI Transport

You can expose the runtime over HTTP without changing your swarm definitions:

pip install "swarmforge[api]"
uvicorn swarmforge.api.fastapi:create_fastapi_app --factory --reload

That app exposes both stateless run endpoints and session-backed endpoints with SSE streaming.

Documentation

• Getting Started
• Create Your First Agent
• Create Your First Multi-Agent Swarm
• Authoring
• Orchestration
• Providers
• API
• Evaluation
• Examples

Source Examples

The repository includes end-to-end example scripts under examples/. Those scripts are useful when you want runnable reference flows for authoring, orchestration, evaluation, provider integration, or FastAPI transport. Provider-backed examples and the local FastAPI example read from .env.example-style settings.

The demo UI under demo-ui/ reads the same root .env for its default API base, provider, and model. Its Vite scripts create .env from .env.example automatically when the file is missing.

Contributing

Core modification, docs development, demo UI work, and PyPI release steps are documented in CONTRIBUTING.md.