swarm-agents 0.1.1

Lightweight framework for LLM agents with tools, hooks, guardrails, and provider routing

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

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://[e-mail]/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_batch(
    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, DialogueRunner, 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 = DialogueRunner()
async for chunk in runner.run_streamed(agent=agent, run_context=run_context, max_turns=10):
    print(chunk)

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 AgentRunner and advisory in DialogueRunner.

See:

• swarm.hooks
• swarm.guardrails
• swarm.runners.agent
• swarm.runners.dialogue

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.