Skip to content

Istos

Istos Documentation

Istos Framework

Decorator-first apps on Eclipse Zenoh — RPC, streaming replies, duplex channels, pub/sub, durable messaging, work queues, and an optional HTTP / SSE / WebSocket / MCP surface.

Python License Status

uv pip install istos
istos new my-service && cd my-service && python main.py

Unauthenticated by default

Default config is Zenoh peer mode, multicast scouting, no TLS. Anyone on the local fabric can hit your handlers. Before production: TLS, an authorizer, and usually require_auth=True. See Security and Authorization.

Who is this for?

Use Istos when… Look elsewhere when…
Your services talk over Zenoh (edge, robotics, agents) Kafka/NATS is the log you operate and keep forever
You want RPC + pub/sub in one Python process The product is a REST API (put FastAPI in front; call Istos via http_port if you still want the fabric)
Peer or router-mediated messaging fits You only ever need HTTP and never a fabric

The mental model

graph LR
    A["@handle & @query"] -->|1-to-1| B["RPC"]
    S["@stream"] -->|1-to-1 chunks| T["Streaming reply"]
    CH["@channel"] -->|duplex| U["Agent session"]
    C["@publish & @subscribe"] -->|1-to-N| D["Events"]
    Q["@worker & enqueue"] -->|1-of-N, once| V["Work queue / jobs"]
    E["@on_liveliness"] -->|presence| F["Who is up"]
Decorators What it does
RPC @handle, @query one request, one reply
Streaming RPC @stream, stream_query one request, many chunks (optional SSE)
Duplex @channel, open_channel interactive agent sessions (optional WebSocket)
Events @publish, @subscribe fire-and-forget
Work queues @worker, @queue, enqueue one worker per job, leases, retries, dead-letter
Scheduling schedule(..., every_s=/cron=) periodic jobs on an interval or cron
Discovery liveliness + .istos/capabilities who's up, what they expose
Durability durable=True, persist="s3://…", replay(), storage plugins replay / survive crashes / idempotency
HTTP http_port, http= / ws=, ASGI co-host, MCP JSON + SSE + WebSocket + tools for outside callers
Glue Depends, middleware, authorizer the usual

Quick example

from contextlib import asynccontextmanager
from istos import Istos

istos = Istos()

@istos.handle("robot/move")
async def move(distance: int, speed: str = "normal"):
    return {"status": "success", "distance": distance, "speed": speed}

@istos.subscribe("drone/telemetry")
async def on_telemetry(data):
    print(f"Received: {data}")

@istos.publish("drone/status")
async def broadcast_status():
    return {"status": "online", "uptime": 999}

@istos.query("robot/move")
async def query_move(result):
    return result

@asynccontextmanager
async def on_start(app):
    # Need an open session — lifespan or another handler
    print(await query_move(distance=10, speed="fast"))
    await broadcast_status()
    yield

istos.lifespan = on_start

if __name__ == "__main__":
    istos.serve_docs(web_port=8080)  # AsyncAPI UI
    istos.run()

Learning path

  1. Installation
  2. Getting Started
  3. RPC (@handle / @query / @stream)
  4. Pub/Sub
  5. Work queues (@worker / @queue / enqueue)
  6. Durable messaging
  7. HTTP gateway
  8. Security · Authorization
  9. Deployment

Other guides: capabilities, liveliness, retry, validation, DI, databases, middleware, observability, storage, architecture health, testing, CLI.

Recipes if you want copy-paste starters.

Feature map

Feature Guide API
@handle / @query RPC Handler, Query
@stream RPC, HTTP Stream
@channel / open_channel Channels Channel
@publish / @subscribe Pub/Sub Publish, Subscribe
@worker / @queue / enqueue (leases, DLQ, chords) Work Queues Work Queue
schedule(every_s=/cron=) Work Queues Work Queue
Durable + S3 persist / replay() Durable messaging Durable, Persist
Liveliness Liveliness Liveliness
Capabilities Capabilities Istos
HTTP / SSE / WS / MCP / ASGI HTTP Gateway, MCP Istos, ASGI, MCP
Validation Validation Validation
Retry Retry Retry
TLS / authz Security, Authorization Config, Authz
Depends DI Depends
Middleware / errors Middleware Middleware, Errors
Health / metrics / OTel Observability Health, Metrics, Tracing
Storage plugins Storage Storage
App databases Application DBs Databases
Serialization DI Serialization
AsyncAPI Getting Started AsyncAPI
Routers DI IstosRouter
Test client Testing TestClient
Architecture fitness (istos analyze) Architecture Health Fitness
CLI CLI CLI

Built-in endpoints

Key Purpose Flag
.istos/health liveness enable_health=True
.istos/ready readiness enable_health=True
.istos/metrics Prometheus text enable_metrics=True
.istos/capabilities what this node exposes enable_discovery=True
.istos/docs AsyncAPI serve_docs(...)

With http_port set you also get GET /livez, /readyz, /metrics, plus any http= routes. See HTTP Gateway.

Built-ins inherit the app-wide authorizer.

Optional extras

uv pip install "istos[redis]"
uv pip install "istos[sqlalchemy]"   # + your async driver
uv pip install "istos[s3]"           # persist streams to S3/MinIO
uv pip install "istos[jwt]"
uv pip install "istos[otel]"
uv pip install "istos[all]"          # redis + sqlalchemy + s3 + jwt + otel
uv pip install "istos[dev]"

Before you ship

  • [ ] TLS + Zenoh credentials (mode=client, explicit endpoints)
  • [ ] require_auth=True and an authorizer (or deliberate Public opt-outs)
  • [ ] json_logs=True; health + metrics (OTel if you use it)
  • [ ] Redis or SQLAlchemy ledger if you run more than one process
  • [ ] HTTP probes (/livez, /readyz) when http_port is set
  • [ ] Treat IstosSecurityWarning as an error in CI

More: Deployment · Security · Production recipe

API index