criticalbridge 1.0.1

Python SDK for building governed external agents on CriticalBridge — boots the conformance sidecar, dispatches A2A tasks to your handlers, auto-emits decision lineage.

criticalbridge

Python SDK for building governed external agents on CriticalBridge. Boots the conformance sidecar in-process via subprocess, dispatches A2A /tasks to your handlers, and auto-emits decision lineage to the Brain. The Python side of the integration described in the External Agent Integration Guide.

Install

pip install criticalbridge

v0.1 requires Node.js (>=20) on PATH — the SDK spawns the TypeScript sidecar (@criticalbridge/conformance-sidecar) as a subprocess via npx. v0.2 will ship a standalone binary that removes this dependency.

Hello-world agent

import os
from criticalbridge import ManagedAgent

agent = ManagedAgent(
    brain_base_url="https://api.criticalbridge.ai",
    enrollment_token=os.environ["CB_ENROLLMENT_TOKEN"],
    entity_id="acme-corp",
    agent_id="customer-support-langchain-v2",
    vendor="langchain",
    public_url="https://sidecar.acme.example.com",   # HTTPS, what the Brain dispatches to
)

agent.start()  # boots the sidecar; blocks until /healthz green

@agent.handler("customer.read")
def lookup_customer(request, context):
    # business logic — call your model, your CRM, whatever
    customer = {"id": request["customer_id"], "name": "Sample"}
    return agent.respond(
        action="return_customer_info",
        payload=customer,
        model_invocation={"model": "gpt-4o", "promptTokens": 412, "completionTokens": 87},
        risk_tier="low",
    )

agent.serve()  # blocks until SIGINT

What the SDK gives you

• Sidecar lifecycle — spawn, wait-for-ready, graceful teardown on SIGINT / atexit
• /tasks dispatcher — local HTTP server, routes A2A envelopes to the right @handler
• respond() — builds the A2A response AND queues a decision-lineage emit (NAIC V2 wire shape via cb-conformance)
• Live session — agent.session reads the registration JWT fresh from the sidecar each time (so E2 token refreshes are picked up without a restart)

Configuration

ManagedAgent(...) accepts these kwargs (validated at construction time; bad config raises BadConfigError before any I/O):

Field	Required	Notes
brain_base_url	yes	e.g. https://api.criticalbridge.ai
entity_id	yes	Your CriticalBridge entity
enrollment_token	yes	One-time token from Portal /dashboard/managed-agents/enrollment-tokens
agent_id	yes	Durable per-agent ID within the entity
vendor	no	One of langchain / autogen / crewai / semantic-kernel / haystack / other. Default other.
public_url	no	HTTPS URL the Brain dispatches to the sidecar. Required for production deploys.
sidecar_port	no	Sidecar's listen port (default 8080)
local_port	no	Local /tasks server port (default ephemeral)
display_name	no	Label in the Portal fleet view
capability_tags	no	List of capability strings

Logging

import logging
logging.getLogger("criticalbridge").setLevel(logging.DEBUG)         # everything
logging.getLogger("criticalbridge.sidecar").setLevel(logging.WARNING)  # quiet sidecar passthrough

Set CB_LOG_LEVEL=DEBUG in the environment as a shortcut.

License

Apache-2.0 © Frank Burkitt and CriticalBridge.AI contributors.