Swarm Agents

swarm-agents is a standalone Python package for building LLM agents with:

• provider-agnostic model routing
• function/tool execution
• lifecycle hooks
• input/output guardrails
• streaming dialogue execution
• autonomous batch execution

The import path is swarm.

Package Docs

• Architecture: docs/architecture.md
• Error handling: docs/error-handling.md
• Durable runtime: docs/runtime.md

Releases

Versions are derived from Git tags. A pushed vX.Y.Z tag creates:

• a GitHub Release with generated notes and attached build artifacts
• a PyPI publish job via Trusted Publishing

Install

Base package:

pip install swarm-agents

With optional image helpers:

pip install "swarm-agents[image]"

From Git:

pip install "swarm-agents[dev] @ git+ssh://git@github.com/logarith-ms/swarm-agents.git@vX.Y.Z"

PyPI is the primary distribution channel. Git installs remain useful for pinned previews or local validation.

Quick Start

Swarm is the beginner-facing entry point. It handles provider configuration and model-router wiring for you.

from swarm import Agent, Swarm

client = Swarm()  # loads config from environment via ConfigLoader.load()

agent = Agent(
    name="Assistant",
    model="openai/gpt-4o",
    instructions="You are a helpful assistant.",
)

async for chunk in client.run_streamed(
    agent,
    deps=your_context,
    stream_manager=your_stream_manager,
    message_store=your_message_store,
    retry_policy=your_retry_policy,
    max_turns=10,
):
    print(chunk)

Batch execution:

from swarm import Agent, Swarm

client = Swarm(
    config={
        "openai_api_key": "...",
        "openai_organization": "...",
        "openai_project": "...",
        "anthropic_api_key": "...",
        "google_api_key": "...",
        "vertex_project": "...",
        "vertex_location": "us-east1",
    }
)

agent = Agent(
    name="Planner",
    model="openai/gpt-4o",
    instructions="Work through the task step by step.",
)

result = await client.run(
    agent,
    deps=your_context,
    stream_manager=your_stream_manager,
    message_store=your_message_store,
    retry_policy=your_retry_policy,
    max_turns=5,
)
print(result.signal)

What The Facade Does

Swarm(...) removes only the provider/router wiring:

• ConfigLoader.load()
• ProviderFactory(config=...)
• ModelRouter(factory=...)
• RunContext(..., router=...)

You still provide implementations for the runtime protocols:

• ContextProtocol
• StreamManagerProtocol
• MessageStoreProtocol
• optionally RetryPolicyProtocol

Those implementations are application-specific and are intentionally not bundled as defaults in this package.

Model Fallback

Agent.model supports either:

• a single model string
• a list of models in fallback order

Example:

from swarm import Agent

agent = Agent(
    name="FallbackAgent",
    model=[
        "openai/gpt-4o",
        "anthropic/claude-3-5-sonnet",
        "google/gemini-2.0-flash",
    ],
    instructions="Try providers in order until one succeeds.",
)

The router will try each model in sequence until one succeeds or all fail.

Provider Configuration

ConfigLoader.load() reads these environment variables and maps them to the config keys expected by ProviderFactory:

• OPENAI_API_KEY
• OPENAI_ORGANIZATION
• OPENAI_PROJECT
• ANTHROPIC_API_KEY
• GOOGLE_API_KEY
• VERTEX_PROJECT_ID -> vertex_project
• VERTEX_LOCATION -> vertex_location

Advanced / Explicit Wiring

If you want full control, keep using the manual path directly:

from swarm import Agent, ConfigLoader, Runner, ModelRouter, ProviderFactory
from swarm.context import RunContext

config = ConfigLoader.load()
router = ModelRouter(factory=ProviderFactory(config=config))

agent = Agent(
    name="Assistant",
    model="openai/gpt-4o",
    instructions="You are a helpful assistant.",
)

run_context = RunContext(
    deps=your_context,
    stream_manager=your_stream_manager,
    message_store=your_message_store,
    router=router,
    retry_policy=your_retry_policy,
)

runner = Runner()
async for chunk in runner.run_streamed(agent=agent, run_context=run_context, max_turns=10):
    print(chunk)

Model Prompt Context

RunContext.model_context optionally projects the merged context used in the system prompt. Its signature is Callable[[dict[str, Any]], Mapping[str, Any]] | None, defaulting to None.

from collections.abc import Mapping
from typing import Any

def project_context(full_context: dict[str, Any]) -> Mapping[str, Any]:
    return {"phase": full_context.get("phase", "ready")}

run_context = RunContext(
    deps=your_context,
    stream_manager=your_stream_manager,
    message_store=your_message_store,
    router=router,
    model_context=project_context,
)

run_context.model_context_variables returns a fresh dict. The callback receives a deep copy of deps.current_context_variables merged with run_context_variables (run-local values win). An empty mapping omits the context block; invalid return types and callback errors propagate without falling back to full context. Without a callback, the full merged context is rendered as before.

Both runner modes refresh the prompt between continuing tool turns; streaming handoffs also use the projection. Tools, hooks, guardrails, run/replay persistence, budgets, and checkpoints keep their existing full-state behavior. This hook only controls the rendered context block, not agent instruction callbacks, history, or tool output. Context, overlays, and agent instructions retain their existing order.

Optional Image Support

swarm.util.image_to_base64(...) is available behind the image extra.

If you call it without installing the extra, it raises an ImportError with the install command:

pip install "swarm-agents[image]"

Hooks And Guardrails

• Hooks are observational and never alter execution flow.
• Input guardrails run before the first LLM call.
• Output guardrails are blocking in Runner and advisory in Runner.

See:

• swarm.hooks
• swarm.guardrails
• swarm.runners.runner
• swarm.runners.runner

Development

pip install -e ".[dev]"
ruff format --check .
ruff check .
mypy swarm
pytest tests -q
python -m build

Maintainer Release Flow

1. Merge conventional commits to main.
2. Wait for CI to pass and Auto Tag to create the next semantic vX.Y.Z tag.
3. The pushed tag triggers both:
   • Release for the GitHub Release page and attached wheel/sdist
   • Publish PyPI for Trusted Publishing to PyPI
4. Verify the new version on:
   • https://github.com/logarith-ms/swarm-agents/releases
   • https://pypi.org/project/swarm-agents/

Notes

• swarm_task/ is intentionally not part of this package.
• swarm/ERROR_HANDLING.md is intentionally not promoted into the package root.