Resonate HTTP Service Design Python

Resonate HTTP Service Design — Python

Overview

Route handlers in a Python web framework (FastAPI, Flask, Django) are the entrypoint for durable workflows; they start or await invocations via the Resonate client. Actual business logic runs in worker processes that register durable functions. Downstream services (a DB worker, a search worker) expose their own durable functions and are invoked via RPC.

This skill covers the shape of that split: what runs where, how routes interact with workers, and how webhooks resolve durable promises.

Architecture model

 client → HTTP routes (FastAPI/Flask)
            → Resonate client (r.run / r.rpc / r.promises.resolve)
                → worker group A (business logic)
                    → ctx.rpc → worker group B (specialized, e.g. DB)

• HTTP server: FastAPI/Flask/Django app handling routes; maps HTTP requests to Resonate invocations
• Worker service: a process (same or different host) with r.register(...)-ed functions; connects to the Resonate server
• Resonate server: durable promise store + coordination hub (Rust v0.9.x, single port 8001)

Rules

• Route handlers are ephemeral. Use Client APIs (r.run, r.rpc, r.promises.*). Never use Context APIs in a route handler.
• Durable functions run in workers. async def with await ctx.run/rpc + r.register(fn).
• All external effects (DB, HTTP, filesystem) run inside ctx.run envelopes — never at the top of a durable function, never at module import time.
• Stable invocation IDs. Derive from request inputs (f"order:{order_id}") — not uuid.uuid4() at route-handler time unless you persist the ID in the response so the client can poll.

Route design patterns

1. Submit-and-poll (async HTTP)

Client submits a job, gets an ID, polls for status:

from __future__ import annotations
import os, time
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from resonate.resonate import Resonate

app = FastAPI()
r = Resonate(url=os.environ.get("RESONATE_URL", "http://localhost:8001"))


class JobSubmission(BaseModel):
    order_id: str
    items: list[str]


@app.post("/jobs")
async def submit_job(body: JobSubmission) -> dict:
    job_id = f"order:{body.order_id}"
    # Fire and forget — workflow runs durably in the background
    r.options(target="order-workers").rpc(job_id, "process_order", body.order_id, body.items)
    return {"job_id": job_id, "status": "started"}


@app.get("/jobs/{job_id}")
async def get_job(job_id: str) -> dict:
    try:
        record = await r.promises.get(job_id)
        state = record.state if hasattr(record, "state") else "pending"
        if state == "resolved":
            return {"job_id": job_id, "status": "done", "result": record.value}
        elif state == "rejected":
            return {"job_id": job_id, "status": "failed", "reason": record.value}
        else:
            return {"job_id": job_id, "status": "running"}
    except Exception:
        raise HTTPException(status_code=404, detail="job not found")

2. Submit-and-callback (webhook resolves promise)

Client submits; workflow blocks on ctx.promise(); external system resolves via webhook:

from resonate.types import Value

# Worker
async def approve_then_ship(ctx, order_id: str) -> dict:
    approval = ctx.promise()
    approval_id = await approval.id()
    await ctx.run(notify_approver, order_id, approval_id)
    decision = await approval
    if not decision.get("approved"):
        return {"status": "rejected"}
    await ctx.run(ship_order, order_id)
    return {"status": "shipped"}


# Route — start the workflow
@app.post("/orders/{order_id}/approve-and-ship")
async def start(order_id: str) -> dict:
    r.options(target="order-workers").rpc(
        f"order:{order_id}", "approve_then_ship", order_id
    )
    return {"status": "pending_approval"}


# Route — webhook resolves the durable promise
@app.post("/webhooks/approval/{order_id}")
async def approval_webhook(order_id: str, payload: dict) -> dict:
    await r.promises.resolve(
        f"approval:{order_id}",
        Value(data={"approved": payload.get("approved", False)}),
    )
    return {"ok": True}

3. Synchronous HTTP wrapping a fast workflow

If the workflow completes quickly, block on it and return the result directly:

@app.post("/price-quote")
async def price_quote(body: QuoteRequest) -> dict:
    quote_id = f"quote:{body.sku}:{body.qty}"
    result = await r.options(target="pricing-workers").rpc(
        quote_id,
        "compute_quote",
        body.sku,
        body.qty,
    ).result()
    return {"quote_id": quote_id, "price": result["price"]}

r.rpc(...) returns a handle; calling .result() and awaiting it blocks the route handler. Keep timeouts tight — a slow worker holds the HTTP request open.

Worker-side lifecycle

A worker process:

1. Instantiates Resonate(url=..., group=...)
2. Registers its durable functions
3. Stays alive (via its own event loop or a while True sleep)

# workers/order_worker.py
from __future__ import annotations
import asyncio, os
from typing import TYPE_CHECKING
from resonate.resonate import Resonate

if TYPE_CHECKING:
    from resonate.context import Context

r = Resonate(url=os.environ.get("RESONATE_URL", "http://localhost:8001"), group="order-workers")

async def process_order(ctx: Context, order_id: str, items: list[str]) -> dict:
    await ctx.run(validate_order, order_id, items)
    await ctx.run(charge_payment, order_id)
    await ctx.run(create_shipment, order_id)
    return {"order_id": order_id, "status": "done"}

r.register(process_order)

if __name__ == "__main__":
    async def main() -> None:
        try:
            # Block until Ctrl-C; SDK long-polls for work
            await asyncio.Event().wait()
        finally:
            await r.stop()

    asyncio.run(main())

Separating worker groups

For larger services, split specialization across worker groups:

# db worker group
r_db = Resonate(url=url, group="db")

async def query_orders(ctx: Context, user_id: str) -> list:
    db = ctx.get_dependency(psycopg.Connection)
    return db.execute("SELECT * FROM orders WHERE user_id = %s", (user_id,)).fetchall()

r_db.register(query_orders)

# business worker group
r_biz = Resonate(url=url, group="business")

async def place_order(ctx: Context, user_id: str, cart: list[str]) -> dict:
    # Reach into the DB group via RPC
    history = await ctx.options(target="db").rpc("query_orders", user_id)
    # ... business logic ...
    return {"user_id": user_id, "order_id": "..."}

r_biz.register(place_order)

Each group runs in its own process (or cluster of processes). RPC routes through Resonate's promise store with the server handling fair dispatch.

Combining FastAPI + in-process worker

Small services often put the HTTP server and the worker in the same process:

from __future__ import annotations
import asyncio, os
from fastapi import FastAPI
from resonate.resonate import Resonate

app = FastAPI()
r = Resonate(url=os.environ.get("RESONATE_URL", "http://localhost:8001"))

async def process_order(ctx, order_id: str) -> dict:
    await ctx.run(charge_payment, order_id)
    return {"order_id": order_id, "status": "done"}

r.register(process_order)

@app.post("/orders/{order_id}")
async def start(order_id: str) -> dict:
    r.run(f"order:{order_id}", process_order, order_id)
    return {"status": "started"}

@app.on_event("shutdown")
async def shutdown() -> None:
    await r.stop()

Because r.register registers in the same process, both RPC-routing and in-process r.run(...) work. For higher throughput, move workers to dedicated processes.

Distinct Python idioms

• FastAPI is the leading modern web framework; patterns use FastAPI async def route handlers.
• Pydantic models (BaseModel) for request bodies — idiomatic + validates at parse time.
• r.run(...) is SYNC — it returns a handle immediately; the actual async work is awaiting handle.result(). You can fire-and-forget by not awaiting the handle.
• r.promises.* are all async — always await them in async route handlers.
• Value(data=...) wraps the payload for r.promises.resolve/reject.

Common gotchas

• Don't use Context APIs in a route handler. ctx only exists inside a r.register()-ed function body. A route handler gets a Resonate client, not a Context.
• Don't rely on FastAPI request lifecycle for workflow timing. A durable function can outlive an HTTP request by hours or days. Return a job ID and let the client poll.
• Always await r.stop() on shutdown. Register a shutdown handler (@app.on_event("shutdown")) to drain in-flight work gracefully.
• Don't forget target. When the route handler calls r.rpc(...), specify .options(target="group-name") so the invocation routes to a worker group.

Related skills

• resonate-basic-ephemeral-world-usage-python — Resonate, r.run, r.rpc, r.options(target=), promise management
• resonate-basic-durable-world-usage-python — r.register, ctx.run, ctx.rpc, ctx.promise
• resonate-human-in-the-loop-pattern-python — webhook-driven promise resolution (pattern 2 in this skill, expanded)
• resonate-saga-pattern-python — multi-step workflows suitable for pattern 3 (sync HTTP)