capio

Composable capabilities for Python: resilience, caching, observability, and AI behavior.

A capability runtime, not a decorator library.

Capio is the composable capability layer for Python applications. Apply cross-cutting behavior to functions and methods — retries, caching, timeouts, circuit breaking, rate limiting, tracing, metrics, logging, and more — with one uniform, typed API.

Designed for sync and async Python, generator-friendly, backend-agnostic, and fail-safe by default. The architecture is specified in the RFC documents (RFC-000…033).

---

Table of contents

• Features
• Installation
• Quick start
• How it works
• Capabilities
• Custom capabilities
• Context & events
• CLI
• Documentation
• Development
• License

Features

• 37 batteries-included capabilities — resilience (retry, cache, timeout, circuit_breaker, rate_limit, throttle, debounce), data & auth (audit, auth, validate, serialize, encrypt, mask, dedup), messaging & orchestration (publish, consume, queue, transaction, workflow, cron, compensate, idempotent), AI (llm, llm_cache, semantic_cache, prompt_cache, memory, rag, ingest, tool, agent, guardrails, token_budget, model_router), and observability (trace, metrics, log) — plus a plugin SDK for your own. → every capability and option
• One uniform API — @use.<name>(...) chained form and the equivalent @use(...) composite form. → the use API
• Sync, async, and generators — one decorator works for def, async def, generator, and async-generator functions. → sync + async + generators
• Lazy pipelines — decorating costs microseconds and does no I/O; the pipeline is built on the first call and memoized. → invocation flow
• Fail-safe by default — cache, trace, metrics, and log degrade gracefully when their backend fails; opt in to hard errors with strict mode. → runtime config
• Cancellation-safe — timeouts and cancellations are BaseExceptions, so retry and circuit-breaker never swallow them. → error model
• Event bus — subscribe to structured events (cache.hit, retry.attempt, circuit.open, …) without touching your functions. → all emitted events
• Context injection — read invocation IDs, environment, and deadlines via use.context(). → context & events
• CLI — capio doctor, inspect, graph, and benchmark. → CLI reference

Installation

pip install capio

For development, clone the repo and install with the dev extra:

git clone https://github.com/shashi3070/capio.git
cd capio
pip install -e ".[dev]"

Capio has no third-party runtime dependencies beyond the CLI (Typer).

Quick start

from capio import use

@use.retry(max_attempts=3, backoff="exponential", jitter=True)
@use.cache(ttl="5m")
@use.timeout(seconds=2)
@use.trace()
def search(query: str) -> list[str]:
    ...

Capabilities compose as nested scopes; the decorator written highest runs outermost. The composite form is equivalent (and sorts by priority):

from capio import use

@use(
    retry={"max_attempts": 3, "backoff": "exponential", "jitter": True},
    cache={"ttl": "5m"},
    timeout={"seconds": 2},
    trace=True,
)
def search2(query: str) -> list[str]:
    ...

Async works with the same API:

@use.retry(max_attempts=3)
@use.cache(ttl="30s")
@use.circuit_breaker(failure_threshold=5, reset_timeout="30s")
async def fetch(url: str) -> bytes:
    ...

Every capability and every option is documented in the usage guide.

How it works

Decorating a function attaches metadata (fn.__capio__) and a thin wrapper — nothing else. On the first call, Capio builds the execution pipeline (validating configuration, running the capability lifecycle, resolving backends) and memoizes it. Every call then runs the wrapped function through that pipeline.

Inside the pipeline, capabilities wrap each other like an onion — each runs, delegates to the next via call_next(ctx), and resumes on the way back out. Ordering is outermost-first; the composite form sorts by priority:

Capabilities are fail-safe by default: if the cache, trace, metrics, log, audit, or store backend fails, the invocation proceeds untouched (the failure is emitted as an event). Under strict mode the same failures raise.

A deep, module-by-module code walkthrough lives in the architecture document.

Error handling

Capio raises from the capio.exceptions module. Two rules to remember:

1. Timeouts and cancellations are BaseException subclasses — except Exception will not catch them. Catch CapioTimeoutError or the base CapioCancelledBase explicitly.
2. Everything else derives from CapabilityException (an Exception), with structured attributes capability, code, and extra.

Timeout

A timed-out invocation raises CapioTimeoutError. It subclasses CapioCancelledBase, so catch it explicitly — except Exception will not:

import time
from capio import use
from capio.exceptions import CapioTimeoutError, CapioCancelledBase

@use(timeout={"seconds": 2})
def call():
    print("call def is called!!")
    time.sleep(3)

try:
    call()
except CapioTimeoutError as exc:
    print(f"timed out after {exc.seconds}s")   # exc.seconds == 2.0
except CapioCancelledBase:
    print("cancelled")                         # covers any other capio cancellation

Two things worth knowing about sync timeouts:

1. The call runs to completion first — time.sleep(3) cannot be interrupted at 2s, so CapioTimeoutError is raised after the function returns (3s in). This is the documented cooperative behavior for sync functions (RFC-018 §3.3).
2. For a hard timeout that interrupts at 2s, use an async function — the async path uses asyncio.wait_for and cancels the underlying task:

import asyncio
from capio import use
from capio.exceptions import CapioTimeoutError

@use(timeout={"seconds": 2})
async def call():
    await asyncio.sleep(3)

async def main():
    try:
        await call()
    except CapioTimeoutError:
        print("hard timeout at 2s")   # fired at 2s; the task is cancelled

asyncio.run(main())

Prefer returning a sentinel over raising? Set return_on:

@use.timeout(seconds=1, return_on="timeout")   # returns "timeout" instead of raising
def slow2() -> str:
    ...

return_on and raise_on=True are mutually exclusive (config error).

Retry exhaustion

from capio import use
from capio.exceptions import RetryExhaustedError

@use.retry(max_attempts=3)
def flaky() -> None:
    raise ValueError("boom")

try:
    flaky()
except RetryExhaustedError as exc:
    last_error = exc.__cause__      # the final ValueError
    print(exc.capability)           # "retry"
    print(exc.code)                 # "capio.retry.exhausted"

Use on_final="reraise_original" to re-raise the first failure instead of wrapping it.

Circuit breaker and rate limit

from capio.exceptions import CircuitOpenError, RateLimitExceededError

@use.circuit_breaker(failure_threshold=3)
def call_api() -> dict: ...

@use.rate_limit(limit=1, window="1s")
def tick() -> None: ...

try:
    call_api()
except CircuitOpenError:
    ...  # dependency is unhealthy; fail fast or serve a fallback

try:
    tick()
except RateLimitExceededError as exc:
    print("retry after", exc.retry_after)   # seconds

With use.retry, these two are not retried by default (they are non-retryable unless you explicitly list them in retry_on).

Configuration errors

These are raised at decoration / first-call time and are all CapabilityExceptions: ConfigurationError, UnknownCapabilityError, DuplicateCapabilityError, ConflictingPipelineError, UnsupportedExecutionKindError, CacheKeyError, BackendUnavailableError (strict mode).

Capabilities

Decorator	Capability	Purpose	RFC
use.retry	Retry	Retry failures with backoff + jitter	RFC-017
use.cache	Cache	In-memory cache with TTL	RFC-016
use.timeout	Timeout	Bound execution time	RFC-018
use.circuit_breaker	Circuit Breaker	Fail fast when a dependency is unhealthy	RFC-018
use.rate_limit	Rate Limit	Admission control	RFC-018
use.throttle	Throttle	Bound in-flight concurrency	RFC-018
use.debounce	Debounce	Coalesce rapid calls	RFC-018
use.trace	Trace	Span recording	RFC-019
use.metrics	Metrics	Counters + histograms	RFC-019
use.log	Log	Structured invocation logging	RFC-020
use.audit	Audit	Append-only audit trail	RFC-020
use.auth	Auth	Authentication + scopes/policy	RFC-020
use.validate	Validate	Schema-based input/output checks	RFC-022
use.encrypt	Encrypt	Encrypt sensitive fields	RFC-022
use.mask	Mask	Redact sensitive fields	RFC-022
use.serialize	Serialize	Input/output codec boundary	RFC-022
use.dedup	Dedup	One result for identical calls	RFC-022
use.publish	Publish	Publish payloads to a topic	RFC-023
use.consume	Consume	Dispatch topic messages	RFC-023
use.queue	Queue	Enqueue / process tasks	RFC-023
use.transaction	Transaction	Commit / rollback participants	RFC-023
use.workflow	Workflow	Ordered steps + recovery	RFC-023
use.cron	Cron	Gate calls on a schedule	RFC-023
use.compensate	Compensate	Best-effort rollback actions	RFC-023
use.idempotent	Idempotent	Idempotency-key replay protection	RFC-023
use.llm	LLM	Model provider boundary	RFC-030
use.llm_cache	LLM Cache	Exact-match LLM caching	RFC-030
use.semantic_cache	Semantic Cache	Embedding-similarity caching	RFC-030
use.prompt_cache	Prompt Cache	Provider cache-control markers	RFC-030
use.memory	Memory	Conversational memory load/store	RFC-030
use.rag	RAG	Retrieve + inject context	RFC-030
use.ingest	Ingest	Chunk + index documents	RFC-030
use.tool	Tool	Expose a callable as a model tool	RFC-030
use.agent	Agent	Tool-calling loop	RFC-030
use.guardrails	Guardrails	Input/output safety checks	RFC-030
use.token_budget	Token Budget	Bound input tokens	RFC-030
use.model_router	Model Router	Route requests to a model	RFC-030

See the usage guide for every option of every capability (types, defaults, examples).

Custom capabilities

Write your own capability by subclassing Capability and implementing run. Then register it and use it like a built-in.

1. Simple audit — wrap the call, observe the result:

from capio import Capability, use
from capio.registry import registry
from capio.events import Event

class Audit(Capability):
    name = "audit"
    priority = 550

    def run(self, ctx, call_next):
        result = call_next(ctx)
        ctx.emit(Event("audit.after", {"fn": ctx.fn_name, "result": repr(result)}))
        return result

registry.register(Audit)

@use.audit()
def handler(x: int) -> int:
    return x * 2

2. Async-aware timing — override run_async and await the inner call (implement both run and run_async to support sync and async functions):

import time
from capio import Capability, use
from capio.registry import registry

class Measure(Capability):
    name = "measure"
    priority = 500

    def run(self, ctx, call_next):
        start = time.perf_counter()
        try:
            return call_next(ctx)
        finally:
            print(f"{ctx.fn_name}: {time.perf_counter() - start:.3f}s")

    async def run_async(self, ctx, call_next):
        start = time.perf_counter()
        try:
            return await call_next(ctx)
        finally:
            print(f"{ctx.fn_name}: {time.perf_counter() - start:.3f}s")

registry.register(Measure)

@use.measure()
async def fetch(url: str) -> bytes:
    ...

3. Stateful with config — declare a schema for validated options; state is isolated per decorated function:

from capio import Capability, use
from capio.registry import registry

class RequestCounter(Capability):
    name = "request_counter"
    priority = 900                          # runs outermost
    schema = {
        "log_every": {"type": "int", "default": 100, "min": 1},
        "enable": {"type": "any", "default": None},
    }

    def __init__(self):
        super().__init__()
        self.count = 0

    def run(self, ctx, call_next):
        self.count += 1
        if self.count % self.cfg.log_every == 0:
            print(f"{self.count} calls to {ctx.fn_name}")
        return call_next(ctx)

registry.register(RequestCounter)

@use.request_counter(log_every=2)
def ping() -> bool:
    return True

The full guide — lifecycle hooks (configure / initialize / start / stop), supports, requires_backends, degradation, backends, events, and an end-to-end example — is in the custom capabilities guide.

Context & events

Inject the per-invocation Context into any decorated function:

from capio import use

@use.context()
def handler(ctx):
    return ctx.invocation_id, ctx.env, ctx.strict

Subscribe to capability events:

from capio import default_runtime

default_runtime().event_bus.subscribe("cache.hit", lambda e: print("hit", e.data))

CLI

capio doctor              # environment + plugin smoke check
capio inspect mod.fn      # show a decorated function's pipeline
capio graph mod.fn        # render pipeline order
capio benchmark           # run micro-benchmarks against RFC-027 budgets
capio version             # print version

If your OS blocks pip-generated console scripts, use python -m capio.cli ....

Documentation

• Architecture guide — how each part is built: code walkthrough, snippets, and the invocation flow
• Usage guide — the full manual: every capability and configuration option
• Capability cookbook — a runnable example for each of the 37 capabilities
• Custom capabilities guide — SDK reference, lifecycle, backends, and an end-to-end example
• RFCs — the normative architecture: RFC-000 index, RFC-001 vision, RFC-002 core concepts, RFC-003 use API, RFC-004…024 architecture, RFC-025 errors, RFC-026 security, RFC-027 performance, RFC-028 CLI, RFC-029 testing, RFC-030 AI/agents/LLM/MCP, RFC-031 reference implementation, RFC-032 roadmap, RFC-033 migration/FAQ
• Changelog

Development

pip install -e ".[dev]"
pytest tests/ -v
ruff check .

Status: v1.0.0 — the full reference implementation per RFC-031: all 37 capabilities (resilience, data/auth, messaging/orchestration, AI, observability), 8 built-in backends, 126 tests, ruff clean.

License

MIT — see LICENSE.