🛡️ Tardigrade

Resilience middleware that makes AI agents self-healing.

Retries · Checkpointing · Circuit breakers · Cost budgets · Graceful degradation
Works with LangGraph, CrewAI, OpenAI SDK, or raw API calls.

An agent with 85% accuracy per step has just 20% end-to-end success over 10 steps.
Tardigrade wraps your agent with production-grade resilience so step 7 failing
does not mean starting over from scratch.

Install

pip install tardigrade-ai

# With the real-time dashboard:
pip install tardigrade-ai[dashboard]

Quickstart

from tardigrade import armor, Workflow, RetryConfig, BudgetConfig, StepCostReport

@armor(name="fetch", retry=RetryConfig(max_attempts=3))
def fetch_data(url: str) -> dict:
    return {"result": "..."}, StepCostReport(input_tokens=500, output_tokens=200, model="gpt-5.4")

@armor(name="analyze")
def analyze(data: dict) -> str:
    return "analysis", StepCostReport(input_tokens=2000, output_tokens=1000, model="gpt-5.4")

with Workflow("my_pipeline", budget=BudgetConfig(max_budget_usd=1.00)) as wf:
    data = fetch_data("https://api.example.com")
    result = analyze(data)

If the workflow crashes, rerun it with the same run_id and Tardigrade resumes from the last checkpoint instead of starting from scratch.

Features

Before and After

Examples

Retry a flaky step

from tardigrade import RetryConfig, armor

class RateLimitedError(RuntimeError):
    pass

@armor(retry=RetryConfig(max_attempts=4, retryable_exceptions=(RateLimitedError, TimeoutError)))
def call_provider(prompt: str) -> str:
    ...

Switch to a fallback model when the primary is down

from tardigrade import CircuitBreakerConfig, armor

def call_claude_haiku(prompt: str) -> str:
    return "fallback response"

@armor(circuit_breaker=CircuitBreakerConfig(failure_threshold=3, fallback=call_claude_haiku))
def call_gpt54(prompt: str) -> str:
    ...

Resume a checkpointed workflow

from tardigrade import Workflow, armor

@armor(name="fetch")
def fetch() -> str:
    return "data"

with Workflow("pipeline", run_id="run-001"):
    fetch()

Enforce a workflow budget

from tardigrade import BudgetConfig, StepCostReport, Workflow, armor

@armor(name="summarize")
def summarize(text: str) -> str:
    return "summary", StepCostReport(input_tokens=500, output_tokens=200, model="gpt-5.4")

with Workflow("pipeline", budget=BudgetConfig(max_budget_usd=0.50)):
    summarize("hello")

Return partial results instead of crashing

from tardigrade import DegradationConfig, DegradationPolicy, Workflow, armor

@armor(name="enrich")
def enrich(data: dict) -> dict:
    raise ConnectionError("provider down")

with Workflow("pipeline", degradation=DegradationConfig(policy=DegradationPolicy.COLLECT)) as wf:
    enrich({"id": 1})

assert wf.result is not None
assert wf.result.status == "failed"

Extended runnable examples live in docs/examples.

Dashboard

# CLI
tardigrade dashboard

# Programmatic
from tardigrade import Dashboard

Dashboard().start_in_thread()

Demo

Run the self-contained demo script:

uv run demo/demo_workflow.py

Record the hero GIF when vhs is installed:

vhs demo/hero.tape

API and Project Docs

• API reference
• Contributing guide
• Changelog
• License