Fleets

Run the same agent across hundreds of inputs in parallel. Fleets handle concurrency limits, cost caps, failure policies, and streaming results as items complete.

Fleets require primitivesEnabled: true. Concurrency is capped at the lower of your requested value and your provider's maxConcurrent limit.

Dispatching a fleet

Call session.fleets.dispatch() with an agent definition, an array of inputs, and your concurrency and cost constraints.

fleet.ts

import { Theazo } from 'theazo'

const theazo = new Theazo({ apiKey: 'th_live_...' })

const session = await theazo.sessions.forUser('user_123')

const fleet = await session.fleets.dispatch({
  agent: 'content-writer',
  inputs: [
    { topic: 'AI infrastructure', tone: 'technical' },
    { topic: 'Founder fundraising', tone: 'narrative' },
    { topic: 'YC application tips', tone: 'direct' },
    // ... up to thousands of items
  ],
  concurrency: 10,
  timeout: '300s',            // per-item timeout (duration string, e.g. '300s' or '5m')
  maxCost: { amount: 5000, currency: 'usd' },
  failurePolicy: 'continue',  // 'continue' | 'abort' | 'pause'
})

console.log(fleet.id)            // 'flt_abc123'
console.log(fleet.totalItems)    // 3
console.log(fleet.currentStatus) // 'running'

Dispatch options

The options accepted by fleets.dispatch(). This table is generated from the @theazo/contracts schema — it can't drift from the SDK.

Field	Type	Required	Description
`agent`	`string`	yes	Agent definition name or ID run for each input.
`inputs`	`Record<string, unknown>[]`	yes	One input object per item; the agent runs once per input.
`concurrency`	`number`	—	Max items run at once. Default: 10 (capped at 100).
`timeout`	`string`	—	Per-item timeout as a duration string.
`maxCost`	`object`	—	Cost cap for the whole fleet; it pauses when reached.
`failurePolicy`	`'continue' \| 'abort' \| 'pause'`	—	Default: continue.

Failure policies

continue

Failed items are recorded but the fleet keeps running. Completed items are still available. Use this when partial results are acceptable.

abort

Any failure immediately aborts the fleet, cancelling all pending and in-progress items. Use this when all-or-nothing semantics matter.

pause

Stops processing on failure and preserves all completed items. The fleet enters a paused state and can be manually resumed after investigation. Use this when you want to inspect failures before deciding whether to continue.

Fleet status

Poll fleet.status() to get a summary of progress.

const status = await fleet.status()

console.log(status.status)         // 'dispatching' | 'running' | 'completed' | 'cancelled' | 'cost_limited'
console.log(status.totalItems)     // 500
console.log(status.completedItems) // 234
console.log(status.failedItems)    // 3
console.log(status.totalCost)      // 1870  (integer cents)
console.log(status.progress)       // { completed: 234, failed: 3, pending: 253, running: 10, total: 500 }

Reading results

Paginated results

Fetch completed results in pages. You can filter by status to get only failed items for retry logic.

// First page of completed results
const page = await fleet.results({
  status: 'completed',
  limit: 50,
})

console.log(page.data)     // FleetItemResult[]
console.log(page.hasMore)  // true if more pages remain
console.log(page.cursor)   // pass to next call to page forward

// Fetch next page
const nextPage = await fleet.results({
  status: 'completed',
  limit: 50,
  cursor: page.cursor,
})

// FleetItemResult shape (from fleet.results()):
// {
//   id:       'flti_abc',
//   agentId:  'agt_...',            // null until the item is picked up
//   input:    { topic: 'AI infrastructure', tone: 'technical' },
//   output:   { category: '...' },  // the agent's output object (present once completed)
//   status:   'completed',          // 'pending' | 'running' | 'completed' | 'failed'
//   cost:     12,                   // integer cents
//   duration: 18000,                // milliseconds
//   error:    { message: '...' },   // present only on failed items
// }

Streaming results

Use fleet.stream() to react as work proceeds rather than polling. This is an async iterable of FleetStreamEvents — a discriminated union (switch on event) covering item completions/failures, periodic progress, and terminal events. Events carry item IDs, not bodies — fetch outputs via fleet.results().

stream.ts

const fleet = await session.fleets.dispatch({
  agent: 'classifier',
  inputs: emails,   // array of 500 email objects
  concurrency: 20,
  maxCost: { amount: 10000, currency: 'usd' },
  failurePolicy: 'continue',
})

let totalCost = 0

for await (const evt of fleet.stream()) {
  switch (evt.event) {
    case 'item.completed':
      totalCost += evt.cost   // integer cents
      console.log(`${evt.itemId} done — running total ${totalCost}c`)
      break
    case 'item.failed':
      console.error(`${evt.itemId} failed: ${evt.error}`)
      break
    case 'fleet.progress':
      console.log(`${evt.completed}/${evt.total} complete`)
      break
    case 'fleet.completed':
      console.log('Total cost (cents):', evt.totalCost)
      break
  }
}

// Stream events carry item IDs only — fetch full outputs via fleet.results().

SSE transport

Under the hood, fleet.stream() uses Server-Sent Events (SSE) via the GET /v1/fleets/:id/stream endpoint. You can consume SSE directly if you need lower-level control — for example, from a browser or a non-TypeScript client.

sse-events.ts

// The SSE endpoint emits these event types:

// connected — sent once when the stream opens
// data: { fleetId: 'flt_abc123', status: 'running', totalItems: 500 }

// item.completed — fired each time an item finishes successfully
// data: { itemId: 'flti_...', cost: 12 }   // cost is integer cents
// (fetch the full item body — input/output/duration — via fleet.results())

// item.failed — fired each time an item fails
// data: { itemId: 'flti_...', error: 'timed out' }

// fleet.progress — periodic summary emitted after each batch
// data: { completed: 234, failed: 3, total: 500 }

// fleet.paused — fleet paused by cost cap or a 'pause' failure policy
// data: { reason: 'cost_limit' }

// fleet.completed — all items are done (terminal)
// data: { totalCost: 3241, completedItems: 497, failedItems: 3 }

// fleet.cancelled — fleet was cancelled (terminal)

// heartbeat — keep-alive emitted every ~15s
// data: { t: 1719000000000 }

Consuming the SSE endpoint directly with EventSource:

sse-client.ts

const url = 'https://api.theazo.com/v1/fleets/flt_abc123/stream'
const source = new EventSource(url, {
  headers: { 'Authorization': 'Bearer th_live_...' },
})

source.addEventListener('item.completed', (e) => {
  const item = JSON.parse(e.data)
  console.log(`item ${item.itemId} done — ${item.cost}c`)  // cost is integer cents
})

source.addEventListener('fleet.progress', (e) => {
  const progress = JSON.parse(e.data)
  console.log(`${progress.completed}/${progress.total} complete`)
})

source.addEventListener('fleet.completed', (e) => {
  const summary = JSON.parse(e.data)
  console.log('Fleet finished:', summary)
  source.close()
})

source.addEventListener('fleet.cancelled', () => {
  console.warn('Fleet cancelled')
  source.close()
})

SSE auto-reconnects on disconnect. The server tracks progress via Last-Event-ID — on reconnect, only events after the last received ID are sent. No items are missed and no duplicates are delivered.

Cancelling a fleet

Cancelling stops all pending items immediately. In-flight items run to completion. Results collected so far remain available via fleet.results().

await fleet.cancel()

const status = await fleet.status()
console.log(status.status)  // 'cancelled'

Example: 500-item batch

batch-classify.ts

import { Theazo } from 'theazo'

const theazo = new Theazo({ apiKey: 'th_live_...' })
const session = await theazo.sessions.forUser('user_789', {
  limits: {
    maxCost: { amount: 20000, currency: 'usd', period: 'day' },
  },
})

// Build 500 inputs from your database
const tickets = await db.supportTickets.findMany({ limit: 500 })
const inputs = tickets.map(t => ({
  id: t.id,
  subject: t.subject,
  body: t.body,
}))

// Dispatch with concurrency of 10 and a $50 cost cap
const fleet = await session.fleets.dispatch({
  agent: 'ticket-classifier',
  inputs,
  concurrency: 10,
  timeout: '60s',
  maxCost: { amount: 5000, currency: 'usd' },
  failurePolicy: 'continue',
})

console.log(`Fleet ${fleet.id} dispatched with ${inputs.length} items`)

// Watch progress as items complete (events carry IDs, not bodies)
for await (const evt of fleet.stream()) {
  if (evt.event === 'fleet.progress') console.log(`${evt.completed}/${evt.total} classified`)
  if (evt.event === 'item.failed')    console.error(`item ${evt.itemId} failed: ${evt.error}`)
}

// Once complete, page through results and write outputs to your DB
let cursor: string | undefined
do {
  const page = await fleet.results({ status: 'completed', limit: 100, cursor })
  for (const item of page.data) {
    await db.supportTickets.update({
      where: { id: (item.input as { id: string }).id },
      data:  { category: item.output, classifiedAt: new Date() },
    })
  }
  cursor = page.hasMore ? page.cursor : undefined
} while (cursor)

const final = await fleet.status()
console.log('Completed:', final.completedItems, '/', final.totalItems)
console.log('Failed:',    final.failedItems)
console.log('Total cost (cents):', final.totalCost)  // 3241

API reference

session.fleets.dispatch(opts)Promise<FleetInstance>Dispatch a fleet. Starts agents immediately up to concurrency limit.

fleet.status()Promise<FleetStatus>Summary: status, totalItems, completedItems, failedItems, totalCost (cents), and a progress breakdown.

fleet.results({ status, limit, cursor })Promise<PaginatedList<FleetItemResult>>Paginated item results ({ data, hasMore, cursor }). Filter by status.

fleet.stream()AsyncIterable<FleetStreamEvent>Yields SSE stream events (item.completed/failed, fleet.progress/paused/completed) as work proceeds. Fetch item bodies via results().

fleet.cancel()Promise<void>Stop pending items. In-flight items run to completion.

fleet.currentStatusstringGetter for the last-known fleet status (updated by dispatch/status()/cancel()).