Building Production Agent Systems with SwarmDB and SwarmAuth

Overview

Multi-agent systems face two distinct problems that single-agent frameworks don't address well: agent identity and cross-agent access control. This tutorial walks through building a production agent system using SwarmDB (for agent-native state) and SwarmAuth MCP (for agent-to-agent authorization), with OWASP Agentic Top 10 v2.1 compliance built into the architecture.

By the end, you'll have a multi-agent development pipeline where planner, coder, and reviewer agents each have their own identity, permissions, and audit trail.

Prerequisites

• Docker or a SwarmDB-compatible deployment
• A SwarmAuth server (self-hosted or cloud)
• Python 3.11+ or TypeScript 5+
• Claude Opus 4.7 or GPT-5.6 API access

Setup

Install the dependencies:

pip install swarmdb swarmauth-client anthropic

Start the services:

# SwarmDB
docker run -d -p 6432:6432 swarmdb/server:latest

# SwarmAuth
docker run -d -p 8443:8443 swarmauth/server:latest

Step 1: Define the Agent Identity Graph

Every agent in the system has an identity, permissions, and relationships to other agents:

# agents.yaml
swarm: development-team
members:
  - id: agent:planner
    role: planner
    capabilities: [task-decomposition, prioritization]
  - id: agent:coder
    role: coder
    capabilities: [code-generation, debugging]
  - id: agent:reviewer
    role: reviewer
    capabilities: [code-review, annotation]

delegation_rules:
  - from: agent:planner
    to: agent:coder
    scopes: [github:read, filesystem:read]
  - from: agent:coder
    to: agent:reviewer
    scopes: [github:annotate]

Register the agents with SwarmAuth:

from swarmauth import Client

client = Client(server="https://swarmauth.local:8443", api_key="***")

client.register_agent("agent:planner", capabilities=["task-decomposition"])
client.register_agent("agent:coder", capabilities=["code-generation"])
client.register_agent("agent:reviewer", capabilities=["code-review"])

client.set_policy(policy_path="agents.yaml")

Step 2: Create the Shared State Schema in SwarmDB

Define the state schema that all agents will share:

from swarmdb import SwarmDB

db = SwarmDB.connect(url="swarmdb://localhost:6432")

db.create_schema("dev-pipeline", {
    "task_id": "string",
    "title": "string",
    "status": "enum:planning,coding,review,done",
    "assignee_agent_id": "agent_identity",
    "created_at": "timestamp",
    "metadata": "json",
})

db.create_schema("review-notes", {
    "task_id": "string",
    "note": "string",
    "reviewer_agent_id": "agent_identity",
    "created_at": "timestamp",
})

Step 3: Build the Agents

Each agent is wrapped in an identity-aware client. The SwarmAuth token is attached automatically to every call.

Planner:

from anthropic import Anthropic
from swarmdb import SwarmDB
from swarmauth import Client

planner_db = SwarmDB.connect("swarmdb://localhost:6432").client("agent:planner")
planner_auth = Client(server="https://swarmauth.local:8443")

def plan(requirements: list[str]) -> list[str]:
    task_ids = []
    for req in requirements:
        response = anthropic.messages.create(
            model="claude-opus-4-7",
            system="You are a planning agent. Decompose this requirement into concrete tasks.",
            messages=[{"role": "user", "content": req}],
        )
        for task in parse_tasks(response.content):
            task_id = planner_db.insert("dev-pipeline", {
                "title": task.title,
                "status": "planning",
                "assignee_agent_id": "agent:coder",
                "metadata": {"source_requirement": req},
            })
            task_ids.append(task_id)
    return task_ids

Coder:

coder_db = SwarmDB.connect("swarmdb://localhost:6432").client("agent:coder")

def execute_task(task_id: str):
    task = coder_db.get("dev-pipeline", task_id)
    coder_db.update(task.id, {"status": "coding"})

    # Execute the coding work — this is your actual coding logic
    result = run_code_agent(task.title)

    coder_db.update(task.id, {"status": "review", "metadata": {"result": result}})

    # Delegate to reviewer with a scoped token
    token = auth_client.delegate(
        from_agent="agent:coder",
        to_agent="agent:reviewer",
        scope=["github:annotate"],
        expires_in=3600,
    )
    return token

Reviewer:

reviewer_db = SwarmDB.connect("swarmdb://localhost:6432").client("agent:reviewer")

def review(task_id: str, token):
    task = reviewer_db.get("dev-pipeline", task_id)

    # Verify the token before writing
    verified = auth_client.verify(token.value)
    assert verified.scope == ["github:annotate"]

    review_response = anthropic.messages.create(
        model="claude-opus-4-7",
        system="You are a code review agent. Annotate the code with concrete suggestions.",
        messages=[{"role": "user", "content": task.metadata["result"]}],
    )

    reviewer_db.insert("review-notes", {
        "task_id": task.id,
        "note": review_response.content,
        "reviewer_agent_id": "agent:reviewer",
    })

    reviewer_db.update(task.id, {"status": "done"})

Step 4: Orchestrate the Pipeline

def run_pipeline(requirements: list[str]):
    task_ids = plan(requirements)

    for task_id in task_ids:
        token = execute_task(task_id)
        review(task_id, token)

# Query the final state with writer-provenance filtering
db = SwarmDB.connect("swarmdb://localhost:6432")
done = db.client("agent:planner").query(
    "dev-pipeline",
    where={"status": "done"},
)

reviews = db.client("agent:reviewer").query(
    "review-notes",
    where={"writer_agent_id": "agent:reviewer"},  # A08 mitigation
)

Step 5: Audit and Verify

Every action is logged with agent identity:

audit = db.audit_query(
    since="2026-10-15T00:00:00Z",
    until="2026-10-15T23:59:59Z",
    by_agent=["agent:planner", "agent:coder", "agent:reviewer"],
)

for event in audit:
    print(f"[{event.timestamp}] {event.agent_id} {event.action} {event.resource}")

Verify OWASP v2.1 compliance:

# A07: agent privilege escalation — coder can't access planner's tools
try:
    auth_client.delegate(
        from_agent="agent:coder",
        to_agent="agent:planner",
        scope=["filesystem:write"],
    )
except PolicyViolation as e:
    print(f"Correctly denied: {e.category} — {e.explanation}")

# A08: memory poisoning — filter reads by writer
poisoned_notes = db.client("agent:planner").query(
    "review-notes",
    where={"writer_agent_id": ["agent:reviewer"]},  # Only accept reviewer writes
)

Key Takeaways

• Agent identities are first-class — every agent has its own identity, permissions, and audit trail
• Writer-provenance prevents memory poisoning — reads can filter by which agent wrote them (A08)
• Delegation tokens are scoped — every delegation has explicit scopes and expiry
• Audit logs are per-agent — you can trace every action back to a specific agent
• Compliance is structural — the architecture itself enforces OWASP v2.1 categories

When to Use

• You're building multi-agent systems where agents delegate to each other
• You need OWASP Agentic Top 10 v2.1 compliance
• You want a clean audit trail with per-agent identities
• You need writer-provenance for security-sensitive applications

Resources

• SwarmDB Docs: https://swarmdb.dev/docs
• SwarmAuth Docs: https://swarmauth.dev/docs
• OWASP Agentic Top 10 v2.1: https://owasp.org/www-project-agentic-ai-threats/
• Claude Opus 4.7 API: https://docs.anthropic.com/claude