swarmwire

SwarmWire

Multi-agent orchestration library for TypeScript. Budget-first. Library, not framework.

Coordinate LLM agents through typed, composable patterns — with hard cost limits, conflict resolution, and adaptive routing. Works standalone or with ANCS as its memory backend.

npm install swarmwire

Quick Start

npm install swarmwire

import { Swarm, createProvider } from 'swarmwire'

const swarm = new Swarm({
  providers: [
    createProvider('anthropic', { apiKey: process.env.ANTHROPIC_API_KEY }),
  ],
  budget: { maxCostCents: 100 },
})

// Create agents
const researcher = swarm.agent({
  name: 'researcher',
  role: 'Research topics thoroughly',
  model: { provider: 'anthropic', model: 'claude-sonnet-4-6-20260320' },
})

const writer = swarm.agent({
  name: 'writer',
  role: 'Write clear summaries',
  model: { provider: 'anthropic', model: 'claude-sonnet-4-6-20260320' },
})

// Run
const result = await swarm.run('Research TypeScript ORMs and recommend one', {
  pattern: 'pipeline',
  stages: [
    { name: 'research', agent: researcher },
    { name: 'write', agent: writer },
  ],
  budget: { maxCostCents: 50 },
})

console.log(result.output)
console.log(`Cost: ${result.cost.totalCostCents}c | Tokens: ${result.cost.totalTokens}`)

Why SwarmWire

Problem	SwarmWire's Answer
Token bleeding — AutoGen/CrewAI loops burn money silently	Budget is a hard constraint. Structurally impossible to exceed.
Ceiling trap — easy frameworks (CrewAI) can't scale, powerful ones (LangGraph) are complex from day one	Progressive disclosure: one-liner to full DAG control.
Framework lock-in — Mastra/LangGraph own your app	Library. You call it. No lifecycle hooks, no app structure.
No TypeScript — most frameworks are Python-first	TypeScript-native. Not a port.
Stateless agents — every run starts from zero	Pluggable memory backends. ANCS for persistent cognitive memory.
No cost visibility — no framework tracks cost as a first-class metric	Per-agent, per-provider, per-step cost breakdown on every execution.

Orchestration Patterns

Orchestrator-Worker (default)

Workers run in parallel, synthesizer merges results.

await swarm.run(task, {
  pattern: 'orchestrator-worker',
  agents: [researcher, analyst, synthesizer],
})

Pipeline

Sequential stages — each agent's output feeds the next.

await swarm.run(task, {
  pattern: 'pipeline',
  stages: [
    { name: 'classify', agent: classifier },
    { name: 'process', agent: processor },
    { name: 'review', agent: reviewer },
  ],
})

Map-Reduce

Split input, process chunks in parallel, reduce.

await swarm.run(task, {
  pattern: 'map-reduce',
  mapper: (input) => splitIntoChunks(input, 10),
  worker: analyzerAgent,
  reducer: summarizerAgent,
  maxParallel: 5,
})

Debate

Agents argue positions, a judge resolves.

import { runDebate } from 'swarmwire'

await runDebate(task, {
  pattern: 'debate',
  proponents: [optimist, pessimist],
  judge: judgeAgent,
  rounds: 3,
  convergenceThreshold: 0.85,
}, providers, budget)

Blackboard

Shared state space with iterative refinement.

import { runBlackboard } from 'swarmwire'

await runBlackboard(task, {
  pattern: 'blackboard',
  agents: [dataAgent, modelAgent, vizAgent],
  rounds: 5,
  convergence: (state) => state.merged.qualityScore > 0.9,
}, providers, budget)

Fan-Out

Same input, N agents, all parallel. Promise.allSettled for agents.

import { runFanOut } from 'swarmwire'

const result = await runFanOut(task, {
  agents: [reviewer1, reviewer2, reviewer3],
  input: codeToReview,
  optional: true,  // individual failures don't kill the batch
}, providers, budget)
// result.output = [output1, output2, output3]

Evolving Orchestration

Adaptive agent sequencing that learns from execution traces.

import { EvolvingOrchestrator } from 'swarmwire'

const orch = new EvolvingOrchestrator()
const result = await orch.run(task, {
  agents: [researcher, analyst, coder, reviewer],
  maxRounds: 10,
  explorationRate: 0.15,
}, providers)

Budget Engine

Every operation has a budget. Hard limits, not advisory.

const result = await swarm.run(task, {
  budget: {
    maxTokens: 100_000,      // Hard token cap
    maxCostCents: 150,        // Hard cost cap
    maxLatencyMs: 30_000,     // Wall-clock deadline
    maxAgents: 5,             // Concurrency cap
    warningAt: 0.8,           // Fire event at 80%
  },
})

// Detailed cost breakdown
result.cost.perAgent       // Map<agentName, { tokens, costCents, calls }>
result.cost.perProvider    // Map<providerName, { tokens, costCents, cacheHits }>
result.cost.budgetUsed     // 0-1 fraction consumed

If budget is exhausted mid-execution: running steps complete, no new steps start, best-effort partial result returned.

Agent Templates

Ready-to-use agents with sensible defaults:

import { Swarm, templates } from 'swarmwire'

const swarm = new Swarm({ providers })

const researcher = swarm.agent(templates.researcher())
const reviewer   = swarm.agent(templates.codeReviewer())
const synth      = swarm.agent(templates.synthesizer())
const analyst    = swarm.agent(templates.dataAnalyst())
const tester     = swarm.agent(templates.qaTester())
const writer     = swarm.agent(templates.writer())
const planner    = swarm.agent(templates.planner())

// Override anything
const cheapResearcher = swarm.agent(templates.researcher({
  modelTier: 'cheap',
  maxCostCents: 5,
}))

YAML Workflows

CI/CD-style workflow definitions:

name: research-and-summarize
version: 1.0.0

inputs:
  topic: string
  depth: string

steps:
  - id: research
    type: llm
    agent: researcher
    prompt: "Research: {{ inputs.topic }}"

  - id: summarize
    type: llm
    agent: writer
    prompt: "Summarize findings about {{ inputs.topic }}"
    dependencies: [research]

import { parseWorkflow, compileWorkflow } from 'swarmwire'

const workflow = parseWorkflow(yamlString)
const plan = compileWorkflow(workflow, {
  agents: new Map([['researcher', researcher], ['writer', writer]]),
  inputs: { topic: 'TypeScript ORMs', depth: 'thorough' },
})

const result = await swarm.execute(plan)

Plan → Inspect → Execute

Don't run blind. Preview the plan, modify it, then execute.

const plan = await swarm.plan('Analyze our auth architecture')

console.log(plan.estimatedCost)        // Preview cost
console.log(visualizePlan(plan))       // ASCII DAG

plan.steps[1].agent = alternateAgent   // Swap an agent
plan.steps[2].optional = true          // Make a step optional

const result = await swarm.execute(plan)
console.log(explainExecution(result))  // Full human-readable report
console.log(summarizeExecution(result)) // One-line summary

Provider Infrastructure

Multi-Provider with Failover

import { createProvider, withCircuitBreaker, withFailover, withRateLimit } from 'swarmwire'

const anthropic = withRateLimit(
  withCircuitBreaker(createProvider('anthropic', { apiKey: '...' })),
  { requestsPerMinute: 50 },
)

const openai = withRateLimit(
  withCircuitBreaker(createProvider('openai', { apiKey: '...' })),
  { requestsPerMinute: 60 },
)

// Automatic failover when primary circuit trips
const provider = withFailover([anthropic, openai])

Cost Optimization

import { analyzeCosts } from 'swarmwire'

const recommendations = analyzeCosts(result)
// [
//   { type: 'tier_downgrade', description: 'Agent "researcher" used 500 tokens but cost 15c...', estimatedSavingsCents: 9 },
//   { type: 'caching', description: 'Only 5% cache hit rate...', estimatedSavingsCents: 12 },
// ]

Protocol Support

MCP — Agent-to-Tool

import { loadMcpTools } from 'swarmwire'

const tools = await loadMcpTools('npx @some/mcp-server')
const agent = swarm.agent({ name: 'tooled', role: '...', tools })

A2A — Agent-to-Agent

import { startA2AServer, importA2AAgent } from 'swarmwire'

// Expose your agents
startA2AServer({ port: 8080, agents: [researcher, analyst] })

// Consume external agents
const externalAgent = await importA2AAgent({ url: 'https://partner.api' })
swarm.register(externalAgent)

Memory Backends

Without Memory (default)

Every execution is stateless. Results returned and forgotten.

With ANCS (coming soon)

Persistent cognitive memory with truth tracking, entity graphs, and importance decay.

import { Swarm, ancsMemory } from 'swarmwire'

const swarm = new Swarm({
  providers,
  memory: ancsMemory({
    url: 'http://localhost:3000',
    tenantId: 'my-project',
  }),
})

Observability

Events

swarm.on('step:start', (e) => console.log(`Starting ${e.agentName}`))
swarm.on('step:complete', (e) => console.log(`Done: ${e.durationMs}ms, ${e.costCents}c`))
swarm.on('budget:warning', (e) => console.log(`Budget at ${(e.usage * 100).toFixed(0)}%`))
swarm.on('conflict:detected', (e) => console.log(`Conflict: ${e.conflict}`))

Streaming

for await (const event of swarm.stream(task)) {
  console.log(event.type, event)
}

Execution Reports

import { explainExecution, summarizeExecution } from 'swarmwire'

console.log(summarizeExecution(result))
// [OK] 3/3 steps | 2.1s | 42.70c | 47.8k tokens

console.log(explainExecution(result))
// Full report: steps, cost breakdown, trace, conflicts

SSE Streaming (Web)

Pipe agent execution to HTTP clients via Server-Sent Events. Works with Express, Fastify, Next.js, or native http. See docs/sse-streaming.md for full recipes.

import { sseHeaders, pipeToSSE } from 'swarmwire/transport'

app.get('/api/run', async (req, res) => {
  sseHeaders(res)
  const result = await pipeToSSE(swarm.stream('Analyze codebase'), res)
  res.end()
})

// Client
const source = new EventSource('/api/run')
source.addEventListener('step:complete', (e) => {
  const { agentName, costCents } = JSON.parse(e.data)
  console.log(`${agentName} done: ${costCents}c`)
})
source.addEventListener('result', (e) => {
  console.log('Output:', JSON.parse(e.data).output)
  source.close()
})

MessageBoard (Inter-Agent Communication)

Agents can communicate ad-hoc during execution through a shared MessageBoard, accessible via ctx.board inside any agent's execute() function. This sits alongside the structured DAG data flow and enables direct messages, broadcasts, topic-based channels, and priority signals.

async execute(input: string, ctx: AgentContext) {
  // Read findings from other agents
  const findings = ctx.board.findings()

  // Broadcast a discovery
  ctx.board.post('*', 'Found a critical issue in auth module', {
    type: 'finding',
    priority: 'urgent',
    data: { file: 'auth.ts', line: 42 },
  })

  // Ask a specific agent
  ctx.board.post('security-expert', 'Is this a real vulnerability?', {
    type: 'question',
  })

  // Read inbox
  const messages = ctx.board.inbox()
}

Message types: finding, warning, question, answer, coordination, status, custom. Priorities: normal, high, urgent.

The full MessageBoard class is also available standalone:

import { MessageBoard } from 'swarmwire'

const board = new MessageBoard(10_000) // max messages
board.post('agent-a', '*', 'Hello everyone', { type: 'status' })
board.stats() // { totalMessages, channels, byType, byAgent, byPriority }

Routing Stack

SwarmWire includes a 5-layer cost-optimization routing stack that can cut LLM API spend by 40-85% with minimal quality loss. Each layer works independently or combined. See docs/routing.md for full details.

Layer	Component	What it does
1	SemanticCache	Embeds queries as vectors, returns cached responses on near-duplicate hits (zero cost).
2	LatencyRouter	Picks the fastest model meeting quality/cost constraints via EMA + P95 latency tracking.
3	CascadeRouter	Tries cheapest model first, escalates if quality is below threshold. Bandit learning over time.
4	SpeculativeCascade	Runs N models in parallel, accepts cheapest that passes quality. Trades cost for latency.
5	QueryDecomposer	Breaks multi-part queries into subtasks, routes each to the cheapest model at its complexity tier.

import {
  SemanticCache,
  LatencyRouter,
  CascadeRouter,
  speculativeCascade,
  decomposeQuery,
  executeDecomposed,
  buildModelLadder,
} from 'swarmwire'

Guardrails

Input, output, and tool-level safety checks with fail-fast tripwires. Inspired by OpenAI Agents SDK. Guardrails run in parallel (default, lower latency) or blocking (sequential, safer) mode. A block-severity failure throws GuardrailTripped and cancels execution immediately; warn severity logs and continues. Sanitization guardrails can modify the value in flight.

import {
  piiGuardrail,
  injectionGuardrail,
  hallucinationGuardrail,
  maxLengthGuardrail,
  contentFilter,
} from 'swarmwire'

const agent = swarm.agent({
  name: 'safe-agent',
  role: 'Process user input safely',
  guardrails: {
    input: [piiGuardrail(), injectionGuardrail()],
    output: [hallucinationGuardrail(), maxLengthGuardrail(10_000)],
    toolInput: [contentFilter(['DROP TABLE', 'rm -rf'], 'block')],
  },
})

Built-in guardrails:

Guardrail	Phase	What it checks
piiGuardrail()	input	Emails, SSNs, credit cards, phone numbers
injectionGuardrail()	input	"Ignore previous instructions" and similar injection patterns
hallucinationGuardrail()	output	Hedging markers ("as of my knowledge cutoff", etc.)
maxLengthGuardrail(n)	output	Truncates output exceeding n chars (warn + sanitize)
contentFilter(strings[], severity)	any	Blocks or warns on forbidden substrings

Custom guardrails implement the Guardrail<T> interface with a check(value, context) method.

Evals Framework

Automated quality metrics for agent outputs. Run evals against Record/Replay fixtures in CI/CD -- no LLM calls needed.

import {
  runEvalSuite,
  nonEmpty,
  lengthCheck,
  containsKeywords,
  schemaMatch,
  similarityToExpected,
  noRegression,
  noHallucination,
} from 'swarmwire'

const suite = {
  name: 'research-quality',
  evals: [nonEmpty(), lengthCheck(100, 5000), containsKeywords(['TypeScript', 'ORM']), noHallucination()],
  threshold: 0.8,          // average score must be >= 0.8
  perEvalThreshold: 0.5,   // no individual eval below 0.5
}

const result = await runEvalSuite(suite, input, output, { expected: groundTruth })
// result.passed, result.averageScore, result.failedEvals

Built-in metrics: nonEmpty, lengthCheck, containsKeywords, schemaMatch, similarityToExpected (Jaccard), noRegression (compare to prior run), noHallucination. All return 0-1 scores. runEvalBatch runs a suite against multiple test cases and reports an overall pass/fail.

Record/Replay Testing

Deterministic, zero-cost testing for multi-agent systems. Record real LLM interactions once, then replay them in CI forever -- instant, free, and reproducible. Fuzzy matching handles volatile fields (UUIDs, timestamps).

import { RecordingProvider, ReplayProvider } from 'swarmwire'

// 1. Record: wrap a real provider, run your workflow, save fixtures
const recording = new RecordingProvider(realProvider, './fixtures/research.json')
await swarm.run('Research TypeScript ORMs', { /* uses recording as provider */ })
await recording.save()   // writes fixture file to disk

// 2. Replay: load fixtures, run the same workflow with zero LLM calls
const replay = new ReplayProvider('./fixtures/research.json')
const result = await swarm.run('Research TypeScript ORMs', { /* uses replay */ })

// 3. Assert: combine with evals
const evalResult = await runEvalSuite(suite, input, result.output)
expect(evalResult.passed).toBe(true)

ReplayProvider options: strict (throw on unmatched requests, default true), fallback (a real provider for partial replay), simulatedLatencyMs.

New Providers

Gemini

Uses Google's OpenAI-compatible endpoint. Models: gemini-2.0-flash, gemini-2.5-pro, gemini-2.5-flash.

const gemini = createProvider('gemini', { apiKey: process.env.GOOGLE_API_KEY })

Ollama (local)

Local execution via Ollama's OpenAI-compatible API. Cost is always $0. Default models: llama3.3, qwen3, deepseek-r1.

const ollama = createProvider('ollama')  // defaults to localhost:11434

Generic OpenAI-compatible / LiteLLM

Any unknown provider name falls through to the OpenAI adapter. Works with LiteLLM, vLLM, Azure OpenAI, or any endpoint that speaks the OpenAI chat completions API.

const litellm = createProvider('litellm', {
  baseUrl: 'http://localhost:4000/v1',
  apiKey: process.env.LITELLM_KEY,
})

Approval Gates

Pause execution before a step and wait for human (or programmatic) approval. If no onApproval callback is provided, gates auto-approve.

const plan = await swarm.plan('Deploy to production')

// Add a gate to the deploy step
plan.steps[2].gate = {
  type: 'approval',
  message: 'Approve deployment to prod?',
  timeoutMs: 60_000,
}

const result = await swarm.execute(plan, {
  onApproval: async (gate) => {
    console.log(`[GATE] ${gate.agentName}: ${gate.message}`)
    return userSaidYes ? 'approved' : 'rejected'
  },
})

Dry-Run Cost Projection

Simulate plan execution without calling any LLMs. Returns cost/duration estimates with min/max/likely ranges, per-step breakdowns, parallelism analysis, and a willExceedBudget flag.

import { dryRun } from 'swarmwire'

const plan = await swarm.plan('Analyze codebase')
const projection = dryRun(plan, providers)

console.log(projection.estimatedCost)
// { minCents: 12.5, maxCents: 50.0, likelyCents: 25.0 }
console.log(projection.willExceedBudget)     // true/false
console.log(projection.stepBreakdown)        // per-step cost + duration
console.log(projection.sequentialDepth)      // critical path length

Output Contracts

Schema + semantic validation of agent outputs. Catches syntactically valid but semantically garbage results. Supports Zod schemas, custom validation functions, and configurable failure actions (retry, skip, fallback, escalate).

import { withContract, OutputContract } from 'swarmwire'

const contract: OutputContract<{ summary: string; score: number }> = {
  schema: z.object({ summary: z.string().min(10), score: z.number().min(0).max(1) }),
  validate: async (output) => ({
    valid: output.score > 0.3,
    reason: output.score <= 0.3 ? 'Score too low — likely garbage output' : undefined,
  }),
  onFailure: 'retry',
  maxRetries: 2,
}

const guardedExecute = withContract(agent.execute, contract)

Throws ContractViolationError when retries are exhausted and onFailure is 'retry' or 'escalate'.

Model Cascade on Quality

Per-agent model fallback that escalates to a smarter (more expensive) model when output quality is too low. Different from circuit breaker (which operates at the provider level on errors).

import { chatWithCascade } from 'swarmwire'

const result = await chatWithCascade(request, {
  primary: { provider: 'anthropic', model: 'claude-haiku-4-20250414' },
  fallbacks: [
    { provider: 'anthropic', model: 'claude-sonnet-4-6-20260320', condition: 'quality' },
    { provider: 'openai', model: 'gpt-4o', condition: 'both' },
  ],
  qualityThreshold: 0.6,
  qualityEstimator: myQualityFn,
}, providerMap)

// result.escalated, result.modelUsed, result.modelsAttempted

Differential Execution

Only re-run steps whose inputs changed. Compares a new plan against a previous ExecutionResult, identifies changed/reusable/cascade steps, and carries forward completed outputs.

import { diffPlans, applyPreviousResults } from 'swarmwire'

const diff = diffPlans(newPlan, previousResult)
// diff.changedSteps, diff.reusableSteps, diff.cascadeSteps, diff.savingsFraction

applyPreviousResults(newPlan, previousResult, diff)
const result = await swarm.execute(newPlan)  // skips reusable steps

Structured Output

Force the LLM to respond with valid JSON matching a schema. Available via ctx.llm<T>() inside any agent's execute() function. Works across providers: maps to response_format on OpenAI/Gemini and tool-use forcing on Anthropic.

const agent = swarm.agent({
  name: 'classifier',
  role: 'Classify support tickets',
  async execute(input: string, ctx: AgentContext) {
    return ctx.llm<{ category: string; priority: number }>(input, {
      responseFormat: {
        type: 'json_schema',
        schema: {
          type: 'object',
          properties: {
            category: { type: 'string', enum: ['bug', 'feature', 'question'] },
            priority: { type: 'number', minimum: 1, maximum: 5 },
          },
          required: ['category', 'priority'],
        },
      },
    })
  },
})

Plugin System

Extend SwarmWire with third-party providers, agents, guardrails, evals, tools, and middleware. See docs/plugins.md for full guide.

import { Swarm, definePlugin, piiGuardrail, noHallucination } from 'swarmwire'

const securityPlugin = definePlugin({
  name: '@myco/security',
  version: '1.0.0',
  guardrails: {
    input: [piiGuardrail()],
    output: [contentFilter(['internal-only'], 'block')],
  },
  evals: [noHallucination()],
  middleware: {
    async beforeExecute(agentName, input) {
      console.log(`[audit] ${agentName} starting`)
      return input
    },
  },
})

const swarm = new Swarm({ providers })
await swarm.use(securityPlugin)

Architecture

User Code
    |
    v
  Swarm  ──────────────────────────────────────────────────────────
    |          |           |            |          |          |
  Planner   Router     Executor      Budget    Patterns   Guardrails
  (DAG)    (cascade    (parallel      Engine   (orch-wkr   (input
   |       semantic     runner        (hard     pipeline    output
  Scorer   cache        dry-run       limits)   map-reduce  tool)
   |       latency      diff-exec       |       debate
  Query    specul.)     checkpoint      |       blackboard
  Decomp.  model-       approval        |       evolving)
           cascade      gates)          |
    |          |           |            |          |          |
    v          v           v            v          v          v
  Providers     MessageBoard    MCP Tools     Memory    Testing
  (Anthropic    (inter-agent    (any server)  (ANCS    (Record/Replay
   OpenAI       communication)                or       Evals
   Gemini                                     custom)  Contracts)
   Ollama
   LiteLLM/generic
   +circuit breaker
   +rate limiter
   +failover)

Project Stats

86 modules | 34 test files | 299 tests | 7 agent templates | 8 docs

Documentation

Guide	What it covers
Routing Stack	5-layer cost optimization, cascade routing, semantic cache, OTEL export
Eval Workflow	Record → Replay → Eval → CI pipeline
SSE Streaming	Express, Fastify, Next.js, React recipes
Conflict Resolution	Detection algorithms, resolution strategies
Persistence	Checkpoint/resume, differential execution, state management
Adapters	Claude Agent SDK, FileBoard, CognitiveVault
Plugins	Plugin interface, middleware, publishing
CognitiveVault	CV-backed inter-agent messaging

License

Free for open source projects, small businesses (under EUR 25,000/year), and personal/educational use. Commercial use above that threshold requires a paid license. See LICENSE.