service-bridge 1.9.0.dev52

ServiceBridge SDK for Python — RPC, events, workflows, jobs, and observability without a mesh

service-bridge

The unified runtime for microservices — RPC, events, workflows, jobs, and observability without a mesh.

Python SDK for ServiceBridge — direct gRPC between workers with zero proxy hops; durable events, background jobs, long-running workflows, and distributed tracing in one SDK. Full mesh with circuit breakers, auto mTLS, and hot-reload transport config. Node, Python, and Go — one identical API. One Go runtime and PostgreSQL.

┌─────────────────────────────────────────────────────────────────┐
│                    BEFORE: 10 moving parts                      │
│  Istio · Envoy · RabbitMQ · Temporal · Jaeger · Consul ·       │
│  cert-manager · Alertmanager · cron · custom glue              │
└─────────────────────────────────────────────────────────────────┘
                              ↓
┌─────────────────────────────────────────────────────────────────┐
│               AFTER: ServiceBridge + PostgreSQL                 │
│  RPC · Events · Workflows · Jobs · Tracing · mTLS · Dashboard  │
│            One SDK  ·  One runtime  ·  Zero sidecars            │
└─────────────────────────────────────────────────────────────────┘

Table of Contents

• Why ServiceBridge
• Use Cases
• Quick Start
• Worker lifecycle
• Runtime Setup
• End-to-End Example
• Platform Features
• How It Compares
• API Reference
• HTTP Plugins
• Configuration
• Environment Variables
• Error Handling
• When to Use / When Not to Use
• FAQ
• Community and Support
• License

---

Why ServiceBridge

Problem	Without ServiceBridge	With ServiceBridge
Service-to-service calls	Istio/Envoy sidecar proxy per pod	Direct SDK-to-worker gRPC, zero proxy hops
Async messaging	Kafka/RabbitMQ + retry logic + DLQ setup	Built-in durable events with retry, DLQ, replay
Background jobs	Celery + Redis + cron daemon	Built-in cron and delayed jobs
Workflow orchestration	Temporal/Conductor cluster + persistence	Built-in DAG workflows
Distributed tracing	Jaeger/Tempo + OTEL collector + dashboards	Built-in traces + realtime UI
Service discovery	Consul/etcd + DNS glue	Built-in registry + health-aware balancing
mTLS	cert-manager + Vault PKI	Auto-provisioned certs from service key

Result: 10 tools → 1 runtime. One Go binary + PostgreSQL replaces the entire stack.

---

Use Cases

Microservice communication — Replace sidecar mesh with direct RPC calls. Get sub-millisecond overhead instead of double proxy hop latency.

Event-driven architecture — Publish durable events with fan-out, retries, DLQ, idempotency, and server-side filtering. No broker infrastructure to manage.

Background job scheduling — Cron jobs, delayed execution, and job-triggered workflows in a single API. No Redis, no Celery, no separate queue workers.

Saga / distributed transactions — DAG workflows with typed steps (rpc, event, event_wait, sleep, child workflow). Compensations and rollbacks via workflow step dependencies.

AI agent orchestration — Stream LLM tokens via realtime trace streams with replay. Orchestrate multi-step AI pipelines as workflows.

Full-stack observability — Every RPC call, event delivery, workflow step, and HTTP request traced automatically. One timeline, one dashboard. Prometheus metrics and Loki-compatible log API included.

---

Quick Start

1. Install

pip install service-bridge

For HTTP middleware:

pip install service-bridge[fastapi]
# or
pip install service-bridge[flask]

2. Create a worker (service that handles calls)

import asyncio
from service_bridge import ServiceBridge

sb = ServiceBridge("localhost:14445", "your-service-key")

@sb.rpc.handle("payment.charge")
async def charge(payload: dict) -> dict:
    return {"ok": True, "tx_id": f"tx_{int(asyncio.get_event_loop().time())}"}

asyncio.run(sb.start())

3. Call it from another service

import asyncio
from service_bridge import ServiceBridge

sb = ServiceBridge("localhost:14445", "your-service-key")

async def main():
    result = await sb.rpc.invoke("payment.charge", {
        "order_id": "ord_42",
        "amount": 4990,
    })
    print(result["tx_id"])

asyncio.run(main())

That's it. No broker, no sidecar, no proxy — direct gRPC call between services.

---

Worker lifecycle

Instantiate with ServiceBridge(grpc_url, service_key, opts?) (same constructor shape as NewServiceBridge / new ServiceBridge in other SDKs).

1. Inbound — register handlers with @sb.rpc.handle, @sb.events.handle, etc.
2. Outbound — call rpc.declare, events.declare, workflows.declare, … before start() when strict declarations are required.
3. start() — connects the worker, provisions TLS, and reconciles registrations with the runtime.
4. Reconcile — performed inside the SDK; you do not need a separate public session client for typical services.

---

Runtime Setup

The SDK connects to a ServiceBridge runtime. The fastest way to start:

bash <(curl -fsSL https://servicebridge.dev/install.sh)

This installs ServiceBridge + PostgreSQL via Docker Compose and generates an admin password automatically. After install, the dashboard is at http://localhost:14444 and the gRPC control plane at localhost:14445.

For manual Docker Compose setup, configuration reference, and all runtime environment variables, see the Runtime Setup section in the main SDK README.

---

End-to-End Example

A complete order flow: RPC → Event → Event handler with streaming.

import asyncio
from service_bridge import ServiceBridge, EventContext

# --- Payments service (worker) ---

payments = ServiceBridge("localhost:14445", "key")

@payments.rpc.handle("payment.charge")
async def charge(payload: dict, ctx) -> dict:
    await ctx.stream.write({"status": "charging", "order_id": payload["order_id"]}, "progress")
    # ... charge logic ...
    await ctx.stream.write({"status": "charged"}, "progress")
    return {"ok": True, "tx_id": f"tx_{int(asyncio.get_event_loop().time())}"}

asyncio.run(payments.start())

# --- Orders service (caller + event publisher) ---

orders = ServiceBridge("localhost:14445", "key")

async def process_order():
    charge = await orders.rpc.invoke("payment.charge", {
        "order_id": "ord_42",
        "amount": 4990,
    })

    await orders.events.publish("orders.completed", {
        "order_id": "ord_42",
        "tx_id": charge["tx_id"],
    }, idempotency_key="order:ord_42:completed", headers={"source": "checkout"})

asyncio.run(process_order())

# --- Notifications service (event consumer) ---

notifications = ServiceBridge("localhost:14445", "key")

@notifications.events.handle("orders.*")
async def on_order(payload: dict, ctx: EventContext) -> None:
    if not payload.get("order_id"):
        ctx.reject("missing_order_id")
        return
    await ctx.stream.write({"status": "sending_email"}, "progress")
    # ... send email ...

asyncio.run(notifications.start())

# --- Register workflow DAG (same orders client; run once at startup or from an admin path) ---

from service_bridge import WorkflowStep

async def register_order_fulfillment():
    await orders.workflows.run("order.fulfillment", [
        WorkflowStep(id="reserve",  type="rpc",        service="inventory", ref="stock.reserve"),
        WorkflowStep(id="charge",   type="rpc",        service="payments", ref="payment.charge",     deps=["reserve"]),
        WorkflowStep(id="wait_dlv", type="event_wait", ref="shipping.delivered",  deps=["charge"]),
        WorkflowStep(id="notify",   type="event",      ref="orders.fulfilled",    deps=["wait_dlv"]),
    ])

asyncio.run(register_order_fulfillment())

Every step above — RPC, event publish, event delivery, workflow execution — appears in a single trace timeline in the built-in dashboard.

---

Platform Features

Communication

• Direct RPC — zero-hop gRPC calls with retries, deadlines, and mTLS identity
• Durable Events — fan-out delivery, guaranteed delivery (RabbitMQ-style), at-least-once guarantees, retries, DLQ, replay, idempotency. If a consumer is offline, the message waits in the server-side queue and is dispatched the moment the consumer reconnects — no retry budget consumed while waiting.
• Realtime Streams — live chunks with replay for AI/progress/log streaming
• Service Discovery — automatic endpoint resolution and round-robin balancing
• HTTP Middleware — FastAPI and Flask instrumentation with automatic trace propagation

Orchestration

• Workflows — DAG steps: rpc, event, event_wait, sleep, child workflow
• Jobs — cron, delayed, and workflow-triggered scheduling

Security

• TLS by default — control plane TLS + worker mTLS with gRPC certificate provisioning
• Access Policy — service-level caller/target restrictions and RBAC

Observability

• Unified Tracing — single trace timeline across HTTP, RPC, events, workflows, and jobs
• Metrics — Prometheus-compatible /metrics endpoint (30+ metric families)
• Logs — structured log ingest with Loki-compatible query API; auto-captures Python logging module
• Alerts — runtime alerts for delivery failures, errors, and service health
• Dashboard — realtime web UI for traces, events, workflows, jobs, DLQ, service map, and service keys

---

How It Compares

Concern	Istio + Envoy	Dapr	Temporal + Kafka	ServiceBridge
RPC data path	Sidecar proxy hop	Sidecar/daemon hop	N/A	Direct (proxyless)
Service discovery	K8s control plane	Sidecar placement	External registry	Built-in registry
Durable events + DLQ	External broker	Pub/Sub component	Kafka + consumers	Built-in
Workflow orchestration	External engine	External engine	Built-in	Built-in
Job scheduling	External cron/queue	External scheduler	External scheduler	Built-in
Traces + UI	Jaeger/Tempo + dashboards	OTEL backend + dashboards	Temporal UI	Built-in
Logs for Grafana	Loki + Promtail pipeline	Log pipeline	Log pipeline	Built-in Loki API
Metrics	App/exporter setup	App/exporter setup	Multiple exporters	Built-in /metrics
Security model	Mesh PKI + policy	Deployment-dependent mTLS	Mixed	Service keys + auto mTLS
Operational footprint	Multi-component mesh	Runtime + sidecars	Workflow + broker + DB	One binary + PostgreSQL

---

API Reference

Cross-SDK parity notes

ServiceBridge keeps the core API surface aligned across Node.js, Go, and Python: constructor, rpc / events / jobs / workflows namespaces, streams, start/stop, and typed errors.

Constructor-level defaults for timeout, retries, and retry delay are available across all three SDKs. Differences are naming and language idioms only:

• Python uses decorators for handler registration (@sb.rpc.handle, @sb.events.handle)
• Handler hints are advisory in all SDKs (timeout_ms/retryable/concurrency for RPC, concurrency/prefetch for events)
• Shared start controls: max_in_flight / MaxInFlight / maxInFlight
• Shared start identity/TLS controls: instance_id/InstanceID/instanceId, weight/Weight, tls/TLS
• RPC transport mode (mode="direct"|"proxy") is available in all SDKs
• Python exposes module-level get_trace_context() (returns {"trace_id", "span_id"} or None) and with_trace_context(ctx, fn) for custom integrations; FastAPI/Flask helpers still set trace state on the request. You can also pass trace_id/parent_span_id explicitly on rpc.invoke / events.publish

ServiceBridge(grpc_url, service_key, opts?)

import os
from service_bridge import ServiceBridge, Options

sb = ServiceBridge(
    grpc_url="localhost:14445",
    service_key="sbv2.<id>.<secret>.<ca>",
    opts=Options(
        heartbeat_interval_ms=10_000,
        capture_logs=True,
        queue_max_size=1000,
        queue_overflow="drop-oldest",
        discovery_refresh_ms=10_000,
        timeout_ms=30_000,
        retries=3,
        retry_delay_ms=300,
    ),
)

Service identity is resolved by the runtime from service_key.

Options:

Option	Type	Default	Description
heartbeat_interval_ms	int	10000	Base heartbeat period for worker registrations.
capture_logs	bool	True	Auto-attach a logging handler that forwards logs to ServiceBridge.
queue_max_size	int	1000	Max offline queue size for control-plane writes.
queue_overflow	str	"drop-oldest"	Overflow strategy: "drop-oldest", "drop-newest", "error".
discovery_refresh_ms	int	10000	Discovery refresh period for endpoint updates.
timeout_ms	int	30000	Global default hard timeout per RPC attempt in ms. Per-call timeout_ms overrides.
retries	int	3	Global default retry count. Per-call retries overrides.
retry_delay_ms	int	300	Base retry delay in ms. Exponential backoff: delay * 2^(attempt-1).

Advanced TLS overrides

Option	Type	Default	Description
ca_cert	str	from service_key	Optional control-plane CA override. By default SDK reads CA from sbv2 service key.
worker_tls	WorkerTLSOpts | None	None	Global explicit worker mTLS materials (cert + key required together, ca_cert optional).

WorkerTLSOpts:

from service_bridge import WorkerTLSOpts

WorkerTLSOpts(
    ca_cert="-----BEGIN CERTIFICATE-----...",
    cert="-----BEGIN CERTIFICATE-----...",
    key="-----BEGIN PRIVATE KEY-----...",
    server_name="workers.internal",
)

---

rpc.invoke(fn, payload?, *, retries?, timeout_ms?, retry_delay_ms?, trace_id?, parent_span_id?, mode?)

result = await sb.rpc.invoke(
    "payment.charge",
    {"order_id": "ord_42", "amount": 4990},
    retries=3,
    timeout_ms=5000,
    retry_delay_ms=500,
)

Arguments — fn is the global function name (same string as @sb.rpc.handle, e.g. payment.charge). It must be unique in the catalog and must not contain /. Use dot notation to group methods.

Parameter	Type	Default	Description
fn	str	required	Global function name on the callee.
payload	Any	None	JSON-serialisable payload.
retries	int	from Options (3)	Retry count (0 = no retry).
timeout_ms	int	from Options (30000)	Hard timeout per attempt in ms.
retry_delay_ms	int	from Options (300)	Base retry delay in ms. Exponential backoff: delay * 2^(attempt-1).
trace_id	str	auto	Override trace ID.
parent_span_id	str	auto	Explicit parent span ID for distributed tracing.
mode	str | None	None (direct)	Transport mode. "direct" (default) connects directly to the worker. "proxy" routes through the control plane when direct connection is unavailable.

rpc.invoke() is bounded even when a downstream worker is silent: each attempt has a hard timeout, retries are finite (retries + 1 total attempts), and after the last failed attempt the root RPC span is closed with error.

Retry delay uses exponential backoff: retry_delay_ms * 2^(attempt-1).

# Proxy mode — route through control plane
result = await sb.rpc.invoke("payment.charge", {"order_id": "ord_42"}, mode="proxy")

---

events.publish(topic, payload?, *, idempotency_key?, trace_id?, parent_span_id?, headers?)

message_id = await sb.events.publish(
    "orders.created",
    {"order_id": "ord_42"},
    idempotency_key="order:ord_42",
    headers={"source": "checkout"},
)

Parameter	Type	Default	Description
topic	str	required	Dot-separated event topic.
payload	Any	None	JSON-serialisable payload.
idempotency_key	str	""	Dedup key for safe publishing.
trace_id	str	auto	Override trace ID.
parent_span_id	str	auto	Explicit parent span ID for distributed tracing.
headers	dict[str, str]	None	Custom metadata headers.

Returns message_id (empty string when buffered offline).

events.publish_worker(topic, payload?, *, trace_id?, parent_span_id?, headers?)

Publishes via the open worker session only (call after start()). Lower latency than control-plane events.publish when the worker is connected; fails if no session is active.

---

jobs.run(service, fn, opts) / jobs.run(target, opts)

Two call shapes (same method): await sb.jobs.run(service, fn, opts) schedules an RPC job (opts.via must remain "rpc" — this is the default). await sb.jobs.run(target, opts) schedules a job whose target is a single string; use via="event" or via="workflow" in opts (for these forms via must not be "rpc" — use the three-argument RPC form instead).

from service_bridge import ScheduleOpts

job_id = await sb.jobs.run("payments", "billing.collect", ScheduleOpts(
    cron="0 * * * *",
    timezone="UTC",
    via="rpc",
))

# via="event" or via="workflow": one target string + ScheduleOpts (target_ref format is enforced by the runtime)
job_id = await sb.jobs.run(
    "orders/order.fulfillment",
    ScheduleOpts(cron="0 7 * * *", timezone="UTC", via="workflow"),
)

ScheduleOpts:

Field	Type	Default	Description
cron	str	""	Cron expression.
delay_ms	int	0	One-shot delay in ms.
timezone	str	"UTC"	IANA timezone.
misfire	str	"fire_now"	"fire_now" or "skip".
via	str	"rpc"	"rpc", "event", or "workflow".
retry_policy_json	str	""	Retry policy JSON.

---

workflows.run — register DAG or execute

Python exposes one async method, sb.workflows.run: the type of the second positional argument selects the operation — a list[WorkflowStep] registers or updates a DAG; a str (workflow name) triggers execution on the service given by the first argument (see workflows.run(service, name, input?) — execute below).

workflows.run(name, steps, opts?) — register DAG

Signature (register): await sb.workflows.run(name, steps, opts=None) -> str

Registers (or updates) a workflow definition. Returns the workflow registration id from the runtime (empty string if the request was queued offline while the control plane was unreachable). If opts is None, defaults are used.

from service_bridge import WorkflowStep

await sb.workflows.run("order.fulfillment", [
    WorkflowStep(id="reserve", type="rpc", service="inventory", ref="stock.reserve"),
    WorkflowStep(id="charge", type="rpc", service="payments", ref="payment.charge", deps=["reserve"]),
    WorkflowStep(id="wait_5m", type="sleep", duration_ms=300_000, deps=["charge"]),
    WorkflowStep(id="notify", type="event", ref="orders.fulfilled", deps=["wait_5m"]),
])

With explicit limits:

from service_bridge import ServiceBridge, WorkflowOpts, WorkflowStep

await sb.workflows.run("checkout.flow", steps, WorkflowOpts(step_timeout_ms=60_000))

WorkflowOpts:

@dataclass
class WorkflowOpts:
    state_limit_bytes: int = 0  # default 262144 (256 KB) when zero
    step_timeout_ms: int = 0    # default 30000 (30 s) when zero

Field	Type	Default	Description
state_limit_bytes	int	262144 (256 KB)	Maximum serialized state size in bytes. 0 uses the default.
step_timeout_ms	int	30000 (30 s)	Default per-step timeout in milliseconds. 0 uses the default.

WorkflowStep:

Field	Type	Description
id	str	Unique step identifier in the DAG.
type	str	"rpc", "event", "event_wait", "sleep", "workflow".
service	str	Optional. For workflow steps — owning service; for rpc steps the runtime resolves the target primarily from ref (global function name).
ref	str	Required for rpc, event, event_wait, workflow. For rpc — global function name (same as @sb.rpc.handle), no /. For event/event_wait — topic or pattern. For workflow — child workflow name. Prefer dots, not slashes.
deps	list[str]	Dependencies. Empty/omitted means root step.
if_expr	str	Optional filter expression (step is skipped if false).
timeout_ms	int	Optional timeout for rpc and event_wait steps.
duration_ms	int	Required for sleep steps.

---

workflows.run(service, name, input?) — execute

Signature (execute): await sb.workflows.run(service, name, input=None) -> dict[str, str]

Starts a workflow execution on demand. The workflow must be registered first via workflows.run(name, steps) on that service's worker. An alternative to scheduling via jobs.run(target, ScheduleOpts(via="workflow")) — triggers the execution immediately.

Parameter	Type	Default	Description
service	str	required	Logical service that owns the workflow definition.
name	str	required	Workflow name.
input	Any	None	Optional JSON-serializable input payload.

Returns a dict with traceId (the run id from the control plane). Use traceId with watch_trace() to observe execution in real time. Optional WorkflowExecuteOpts can set trace_id / group_trace_id on the execute request; the response does not echo group_trace_id.

result = await sb.workflows.run("users", "user.onboarding", {"userId": "u_123"})

---

cancel_workflow(trace_id)

await sb.cancel_workflow("trace_01HQ...XYZ")

---

@sb.rpc.handle(fn, *, allowed_callers?, schema?, timeout_ms?, retryable?, concurrency?)

Decorator that registers an RPC handler.

@sb.rpc.handle("charge", allowed_callers=["orders"])
async def charge(payload: dict) -> dict:
    return {"ok": True}

# With full context (stream + trace):
@sb.rpc.handle("ai.generate")
async def generate(payload: dict, ctx) -> dict:
    await ctx.stream.write({"token": "Hello"}, "output")
    await ctx.stream.write({"token": " world"}, "output")
    return {"text": "Hello world"}

timeout_ms, retryable, and concurrency are advisory runtime hints (not locally hard-enforced by SDK).

RpcContext:

Field	Type	Description
trace_id	str	Current trace ID.
span_id	str	Current span ID.
stream	StreamWriter	Real-time stream writer.

StreamWriter:

Method	Signature	Description
write	write(data, key="default") -> None	Append a real-time chunk to the run stream.
end	end(key="") -> None	No-op placeholder for API symmetry (lifecycle managed by runtime).

---

@sb.events.handle(topic, retry_policy_json?, filter_expr?, concurrency?, prefetch?, group_name?)

Decorator that registers an event consumer handler. Use group_name to set a fixed consumer group explicitly; otherwise the SDK derives a group from the service hint and topic (see servicebridge.py).

@sb.events.handle("orders.*")
async def on_order(payload: dict, ctx: EventContext) -> None:
    if not payload.get("order_id"):
        ctx.reject("missing_order_id")
        return
    await ctx.stream.write({"status": "processing"}, "progress")

concurrency and prefetch are advisory runtime hints (not locally hard-enforced by SDK).

EventContext helpers:

• ctx.trace_id — current trace ID
• ctx.span_id — current span ID
• ctx.retry(delay_ms=1000) — request redelivery with delay
• ctx.reject(reason) — reject permanently (moves to DLQ)
• ctx.topic, ctx.group_name, ctx.message_id, ctx.attempt, ctx.headers — delivery metadata (also accessible as ctx.refs-style attributes)
• ctx.stream.write(data, key) — append real-time chunks to trace stream

Delivery guarantee: once a message is accepted by the runtime, delivery to each consumer group is guaranteed. If the consumer is offline, the message waits in the server-side queue and is dispatched automatically the moment the service reconnects — no retry budget is consumed while waiting. After SERVICEBRIDGE_DELIVERY_TTL_DAYS (default 7) days without a consumer, the delivery moves to DLQ with reason delivery_ttl_exceeded.

Duplicate pattern registration throws an error.

---

rpc.declare(fn)

Declares that this service may call the given global RPC function name. Must be called before start().

events.declare(topic)

Declares that this service publishes events to topic. Must be called before start().

workflows.declare(service, name)

Declares that this service triggers workflow name on service. Must be called before start().

---

start(*, host?, max_in_flight?, instance_id?, weight?, tls?)

from service_bridge import WorkerTLSOpts

await sb.start(
    host="localhost",
    max_in_flight=256,
    instance_id="orders-a1",
    weight=10,
    tls=WorkerTLSOpts(cert=CERT_PEM, key=KEY_PEM, ca_cert=CA_PEM),
)
# or
asyncio.run(sb.start())

Blocks until cancelled (for example, by process signal handling). Raises ValueError immediately if no handlers are registered (neither rpc.handle() nor events.handle() have been called). Initial OpenWorkerSession failure fails startup immediately; after startup, reconnect uses exponential backoff (1s → 15s).

Parameter	Type	Default	Description
host	str	"localhost"	Bind host for the worker gRPC server. Use 0.0.0.0 in Docker/Kubernetes so ServiceBridge can reach the worker.
max_in_flight	int	128	Max in-flight runtime-originated commands over OpenWorkerSession.
instance_id	str	""	Stable worker instance identifier override.
weight	int	1	Scheduling/discovery weight hint. Values <=0 normalize to 1.
tls	WorkerTLSOpts | None	None	Per-start worker TLS override (takes precedence over Options.worker_tls).

---

stop()

await sb.stop()

Gracefully shuts down: flushes logs, drains offline queue, closes gRPC channels.

---

watch_trace(trace_id, opts?)

from service_bridge import WatchTraceOpts

async for event in sb.watch_trace(trace_id, WatchTraceOpts(key="output", from_sequence=0)):
    print(event.data)
    if event.done:
        break

trace_id is the stream identifier used by ctx.stream.write(...).

Behavior:

• Auto-reconnect with exponential backoff (500ms → 5000ms) on retryable stream failures.
• Deduplicates by sequence across reconnects.
• Enforces strict JSON for type="chunk" payloads (non-JSON chunk terminates stream with fatal error).
• Enforces internal queue limit 256; overflow is fatal (consumer must drain promptly).

WatchTraceOpts:

Field	Type	Default	Description
key	str	""	Stream key filter ("" = all keys).
from_sequence	int	0	Replay from sequence cursor.

TraceStreamEvent:

Field	Type	Description
sequence	int	Monotonic sequence number.
key	str	Stream lane key.
data	Any	JSON-decoded chunk payload.
done	bool	True when terminal trace_complete arrives.
trace_status	str	Final status on terminal event.

---

HTTP Span Utilities

start_http_span(method, path, trace_id?, parent_span_id?)

span = sb.start_http_span("GET", "/health")
try:
    # handler logic
    span.end(status_code=200)
except Exception as e:
    span.end(error=str(e))

register_http_endpoint(method, route, ...)

sb.register_http_endpoint(
    "GET", "/users/{id}",
    request_schema_json='{"type":"object"}',
    transport="http",
)

Parameter	Type	Default	Description
method	str	required	HTTP method.
route	str	required	Route pattern, e.g. "/users/{id}".
instance_id	str	""	Stable identifier for this process.
endpoint	str	""	Reachable address, e.g. "http://10.0.0.1:8000".
allowed_callers	list[str]	None	Service names allowed to call (RBAC).
request_schema_json	str	""	JSON schema for request validation metadata.
response_schema_json	str	""	JSON schema for response validation metadata.
transport	str	""	Transport label (e.g. "http", "https").

A periodic heartbeat is automatically sent to keep the HTTP endpoint alive in the registry.

---

Trace Utilities

extract_trace_from_headers(headers)

from service_bridge import extract_trace_from_headers
# or
from service_bridge.http_catalog import extract_trace_from_headers

trace_id, parent_span_id = extract_trace_from_headers(request.headers)

Extracts trace context from HTTP headers. Supports W3C traceparent, x-trace-id/x-span-id headers, and synthesizes trace IDs when those headers are absent. Useful for custom HTTP framework integrations (Django, Starlette, etc.).

---

HTTP Plugins

FastAPI (service_bridge.http.fastapi)

pip install service-bridge[fastapi]

from fastapi import FastAPI, Request
from service_bridge import ServiceBridge
from service_bridge.http.fastapi import ServiceBridgeMiddleware, get_client

sb = ServiceBridge("localhost:14445", "key")
app = FastAPI()

app.add_middleware(ServiceBridgeMiddleware, client=sb, exclude_paths=["/health"], auto_register=True)

@app.get("/users/{user_id}")
async def get_user(user_id: str, request: Request):
    client = get_client(request)
    user = await client.rpc.invoke("user.get", {"id": user_id})
    return user

The middleware:

• Injects the SDK client into request.state.servicebridge_client
• Starts/ends HTTP span automatically
• Sets x-trace-id response header
• Auto-registers route patterns in the catalog

---

Flask (service_bridge.http.flask)

pip install service-bridge[flask]

from flask import Flask, g
from service_bridge import ServiceBridge
from service_bridge.http.flask import init_servicebridge

sb = ServiceBridge("localhost:14445", "key")
app = Flask(__name__)
init_servicebridge(app, sb, auto_register=True)

@app.get("/users/<user_id>")
def get_user(user_id):
    # g.service_bridge, g.trace_id, g.span_id available
    return {"id": user_id}

The middleware:

• Injects the SDK client into g.service_bridge
• Provides g.trace_id and g.span_id in handlers
• Starts/ends HTTP span automatically
• Sets x-trace-id response header

---

Configuration

TLS behavior

• Control plane is TLS-only. Trust source is embedded into sbv2 service key by default.
• Embedded/explicit CA PEM is validated with strict x509 parsing.
• Worker serving is always mTLS.
• Worker certificate provisioning is gRPC-only via ProvisionWorkerCertificate.
• mTLS provisioning requires the cryptography package: pip install service-bridge[mtls]

Offline queue behavior

When the control plane is unavailable, SDK queues write operations (event, job, workflow, telemetry writes).

• Queue size: queue_max_size (default: 1000)
• Overflow policy: queue_overflow (default: "drop-oldest")
• Return values for queued writes may be empty strings until flushed

Log capture

By default (capture_logs=True), a ServiceBridgeLogHandler is attached to the root Python logger. All logging.info(), logging.error(), etc. calls are automatically shipped to ServiceBridge with batching (100 records or 500ms flush interval).

---

Environment Variables

The SDK requires values you pass into ServiceBridge(...). Common setup:

Variable	Required	Example	Description
SERVICEBRIDGE_URL	yes	localhost:14445	gRPC control plane URL
SERVICEBRIDGE_SERVICE_KEY	yes	sbv2.<id>.<secret>.<ca>	Service authentication key (sbv2 only)

import os
from service_bridge import ServiceBridge, Options

sb = ServiceBridge(
    os.environ.get("SERVICEBRIDGE_URL", "localhost:14445"),
    os.environ["SERVICEBRIDGE_SERVICE_KEY"],
)

---

Error Handling

ServiceBridgeError carries a stable sb_code (SB_*, enum SB in service_bridge/sdk_errors.py) and an English message (str(e), aligned with SB_MESSAGES). Use sb_code for comparisons and telemetry; grpc_status is set when the failure originated from gRPC.

from service_bridge import ServiceBridge, ServiceBridgeError, SB, SB_MESSAGES

try:
    await sb.rpc.invoke("payment.charge", {"order_id": "ord_1"})
except ServiceBridgeError as e:
    print(e.sb_code, e.component, e.operation, e.severity, e.retryable)
    if e.sb_code == SB.RPC_TARGET_REQUIRED.value:
        print(SB_MESSAGES[SB.RPC_TARGET_REQUIRED.value])
    raise

Field	Type	Description
sb_code	str	Stable SB_* identifier (primary).
component	str	SDK subsystem (for example, "rpc" or "event").
operation	str	Operation that failed.
severity	str	"fatal", "retriable", or "ignorable".
retryable	bool	Whether retry is recommended.

Full code ↔ message table: ../README.md#errors-sb_-codes. Source of truth: service_bridge/sdk_errors.py.

---

When to Use / When Not to Use

ServiceBridge is a good fit when you:

• Have 3+ microservices that need to communicate via RPC, events, or both
• Want RPC + events + workflows + jobs without managing separate infrastructure for each
• Need end-to-end tracing across all communication patterns in one timeline
• Want to eliminate sidecar proxies and reduce operational overhead
• Need durable event delivery with retry, DLQ, and replay without running a broker
• Are building AI/LLM pipelines and need realtime streaming with replay

Consider alternatives when you:

• Run a single monolith with no service decomposition plans
• Need ultra-high-throughput event streaming (100K+ msg/s sustained) — Kafka is purpose-built for this
• Need a full API gateway with rate limiting, auth plugins, and request transformation — use Kong/Envoy Gateway
• Already have a mature Istio/Linkerd mesh and only need traffic management (no events/workflows/jobs)
• Need multi-region event replication — ServiceBridge currently targets single-region deployments

---

Exported Types

Key types available for import:

from service_bridge import (
    ServiceBridge,
    ServiceBridgeError,
    EventContext,
    StreamWriter,
    RpcContext,
    ScheduleOpts,
    WorkflowStep,
    WorkflowOpts,
    WatchTraceOpts,
    TraceStreamEvent,
    RpcFieldDef,
    RpcSchemaOpts,
    Options,
    WorkerTLSOpts,
    ServiceBridgeLogHandler,
    HttpSpan,
    extract_trace_from_headers,
    get_trace_context,
    with_trace_context,
)

Symbol	Kind	Description
ServiceBridge	class	Main SDK client.
ServiceBridgeError	exception	Typed error with sb_code (SB_*), component, operation, severity, retryable.
EventContext	dataclass	Delivery metadata and control methods for event handlers.
StreamWriter	class	Real-time stream writer (write, end).
RpcContext	class	Trace context and stream writer for RPC handlers.
ScheduleOpts	dataclass	Job schedule configuration.
WorkflowStep	dataclass	Single step in a workflow DAG.
WorkflowOpts	dataclass	Workflow-level limits (state_limit_bytes, step_timeout_ms).
WatchTraceOpts	dataclass	Options for watch_trace() — key filter and replay cursor.
TraceStreamEvent	dataclass	Single event from a trace stream.
RpcFieldDef	dataclass	Field definition for inline protobuf schema.
RpcSchemaOpts	dataclass	Schema options for @sb.rpc.handle(...).
Options	dataclass	Constructor-level SDK configuration.
WorkerTLSOpts	dataclass	Explicit worker mTLS materials.
ServiceBridgeLogHandler	class	Python logging.Handler that ships logs to ServiceBridge.
HttpSpan	class	Manual HTTP span for custom tracing.
extract_trace_from_headers	function	Extract (trace_id, parent_span_id) from HTTP headers.
get_trace_context	function	Returns {"trace_id", "span_id"} for the current context, or None.
with_trace_context	function	Runs a callable with a given trace context set (contextvars).

---

FAQ

How does ServiceBridge handle service failures? RPC calls have configurable retries with exponential backoff and hard per-attempt timeouts, so a silent downstream service cannot keep a call pending forever. Events are durable (PostgreSQL-backed) with at-least-once delivery per consumer group. Failed deliveries are retried according to policy, then moved to DLQ. Workflows track step state and can be resumed.

Is there vendor lock-in? ServiceBridge is self-hosted. The runtime is a single Go binary + PostgreSQL. SDK calls map to standard patterns (RPC, pub/sub, cron) — migrating away means replacing SDK calls with equivalent library calls.

How does tracing work without an OTEL collector? The SDK automatically reports trace spans for every RPC call, event publish/delivery, workflow step, and HTTP request. The runtime stores traces in PostgreSQL and serves them via the built-in dashboard and a Loki-compatible API for Grafana integration.

Can I use ServiceBridge alongside existing infrastructure? Yes. You can adopt incrementally — start with RPC between two services, add events later, then workflows. ServiceBridge doesn't require replacing your existing broker or mesh all at once.

What happens when the control plane is down? In-flight direct RPC calls continue working (they go service-to-service, not through the control plane). New discovery lookups, event publishes, and telemetry writes are queued in the SDK offline queue and flushed when the control plane recovers.

What databases does the runtime support? PostgreSQL 16+. The runtime uses PostgreSQL for all persistence: traces, events, workflows, jobs, service registry, and configuration.

---

Low-level session API (V2SessionClient)

Not the primary application API. Use ServiceBridge(...) and start() for normal workers; the SDK owns the session loop and reconcile.

service_bridge.session_v2 implements the Enterprise Channel protocol — the bidi-stream session layer shared symmetrically across Go, Node.js, and Python SDKs — for advanced or custom integrations.

Quick start

from service_bridge.session_v2 import V2Config, V2SessionClient, TransportMode

config = V2Config(
    server_address="localhost:14445",
    instance_id="payments-a1",
    zone="us-east-1a",
    transport_mode=TransportMode.DIRECT,
    max_inflight=128,
)

session = V2SessionClient(config)

# Build Hello message fields for the first connect (or reconnect)
hello_fields = session.get_hello_fields()

# After server sends HelloAck:
from service_bridge.session_v2 import HelloAckV2
ack = HelloAckV2(
    session_id="sess_01",
    resume_token="tok_abc",
    epoch=1,
    resumed=False,
    resume_from_seq=0,
    replayed_commands=0,
    reconciled_results=0,
    heartbeat_interval_ms=10_000,
    heartbeat_timeout_ms=30_000,
    initial_permits=128,
    max_permits=256,
    effective_transport_mode=TransportMode.DIRECT,
)
await session.on_hello_ack(ack)

8-state FSM

State	Description
CONNECTING	Initial state; TCP/TLS handshake in progress
HANDSHAKING	Hello sent; waiting for HelloAck
READY	HelloAck received; idle, no active commands
ACTIVE	At least one command in flight
SUSPENDED	2+ heartbeat misses; reconnect imminent
DRAINING	Graceful shutdown; completing in-flight commands
FENCED	Server sent GOAWAY_FENCED; epoch bump required
CLOSED	Terminal state

Adaptive heartbeat

# Interval adjusts automatically based on EWMA RTT and miss count
interval = session.heartbeat.next_interval_ms()

# Call on pong receipt (rtt_ms = round-trip time)
session.on_pong(rtt_ms=12.5)

# Call on missed pong — returns True when state→SUSPENDED
suspended = session.on_heartbeat_miss()

Credit-based flow control

# Before dispatching a command
ok = await session.on_command_received(seq=42, command_id="cmd_xyz")
if not ok:
    # Backpressure: no permits available
    ...

# After command finishes (releases 1 permit)
await session.on_command_completed(seq=42, command_id="cmd_xyz")

# Server-side grant (PermitGrant frame)
await session.on_permit_grant(additional=10)

# Server-side window resize (FlowControlUpdate frame)
await session.on_flow_control_update(new_window_size=64, reason="load_shedding")

Dynamic ConfigPush

from service_bridge.session_v2 import TransportConfigV2, ServiceTransportOverride, TransportMode

config_push = TransportConfigV2(
    default_mode=TransportMode.DIRECT,
    service_overrides={
        "inventory": ServiceTransportOverride(mode=TransportMode.PROXY),
    },
)
session.on_config_push(config_push)

# Resolve effective transport for any function call
mode = session.resolve_transport_mode("stock.reserve")  # → PROXY
mode = session.resolve_transport_mode("payment.charge")     # → DIRECT

GoawaySignal / drain

# Server-initiated fencing (epoch bump required on next connect)
session.on_goaway(code="GOAWAY_FENCED", reason="epoch_conflict")

# Worker-initiated graceful drain
session.on_drain(reason="sigterm", deadline_ms=30_000)

Reconnect with selective replay

# PositionTrackerV2 tracks seq numbers for bidirectional replay
resume_state = session.position.get_resume_state(
    token=session._resume_token,
    epoch=session._epoch,
)
# resume_state.last_received_seq, last_sent_seq, completed_command_ids
# → pass these fields in the next Hello message

Exported symbols

Class	Description
V2Config	Session configuration (symmetric with Go/Node)
V2SessionClient	Main session client with 8-state FSM
SessionStateV2	FSM state enum
TransportMode	"direct" / "proxy"
ResumeState	Reconnect resume state
HelloAckV2	HelloAck data class
ReconcileRequestV2	Declarative worker registration
TransportConfigV2	Dynamic transport config (ConfigPush)
AdaptiveHeartbeatV2	EWMA RTT heartbeat controller
FlowControlStateV2	Credit-based flow control
BackoffV2	Exponential backoff with full jitter
PositionTrackerV2	Seq-number tracker for selective replay
ConfigPushStateV2	Applied transport config state

---

Community and Support

• Website: servicebridge.dev
• GitHub: github.com/service-bridge
• SDK monorepo: README.md

---

License

Free for non-commercial use. Commercial use requires a separate license. See LICENSE.

Copyright (c) 2026 Eugene Surkov.

---

Keywords

service-bridge · servicebridge · pip install service-bridge · Python SDK · asyncio microservices · RPC · gRPC · event bus · event-driven · distributed tracing · workflow orchestration · background jobs · cron · mTLS · service mesh · service discovery · zero sidecar · Istio alternative · RabbitMQ alternative · Celery alternative · Temporal alternative · Jaeger alternative · PostgreSQL · Docker · Kubernetes · DLQ · dead letter queue · saga · distributed transactions · AI agent orchestration · FastAPI middleware · Flask middleware · HTTP middleware · observability · Prometheus · tracing · service catalog · durable events · retries · idempotency · auto mTLS · runtime dashboard · production ready · microservice communication · Python 3.10+