Multi-Agent & Events

Stitch the decisions from a multi-step agent pipeline together under one workflow, and emit events you can react to as those decisions happen.

correlation events For: platform & governance

Why correlation matters

Before correlation
After correlation

Each agent records its decision independently. Reviewing an escalation means hunting for three separate records and guessing which retrieve, classify, and draft belong to the same ticket. The chain of reasoning is real but invisible.

All three agents run under one workflow_id and register into one root span. The steps trace as a single chain, so a reviewer can reconstruct exactly what context the classifier saw and what the drafter did with it. The chain lives in your traces; attaching the id to the decision records themselves is one line you write (see Limits).

The two surfaces

Surface	What it does	Reach for it when
Correlation (`briefcase.correlation`)	Groups decisions under a shared workflow and propagates context across boundaries	You want the pipeline to read as one accountable chain
Events (`briefcase.events`)	Emits typed signals (`BriefcaseEvent`) you can route as decisions happen	You want to react in real time — alert, page, or trigger follow-up

Install

pip install briefcase-ai[correlation]
pip install briefcase-ai[events]

Correlation

How a workflow threads the pipeline

briefcase_workflow(name, client) is a context manager. Every agent registered inside it shares the same workflow_id, so the steps of a pipeline trace as one unit.

Open a workflow — briefcase_workflow("support_pipeline", client) gives every agent inside it one shared workflow_id.
Retrieve — the retrieval agent registers under that workflow.
Classify — classify_ticket’s agent registers under the same workflow.
Decide — the drafting agent registers under the same workflow.
Review — the three registered agents read back as one retrieve -> classify -> decide chain.

from unittest.mock import Mock
from briefcase.correlation import briefcase_workflow


def retrieve(query):
    return ["doc-12", "doc-44"]


def classify_ticket(docs):
    # call your model here
    return "account_access"


def decide(category):
    return "route_to_support"


client = Mock()  # your Briefcase client; a Mock keeps this example self-contained

with briefcase_workflow("support_pipeline", client) as workflow:
    docs = retrieve("reset my password")
    workflow.register_agent("retriever-1", "retrieve")

    category = classify_ticket(docs)
    workflow.register_agent("classifier-1", "classify")

    action = decide(category)
    workflow.register_agent("decider-1", "decide")

    print(workflow.workflow_id)  # all three share this id
    print(action)                # route_to_support

Agent registration

workflow.register_agent(agent_id, agent_type) records each agent in the workflow, so the chain knows who made which decision.

workflow.register_agent("classifier-1", "classify")

Reading the active workflow

get_current_workflow() returns the workflow bound to the current context, or None outside a workflow block — so you can read it anywhere inside the chain without threading it through call signatures.

from unittest.mock import Mock
from briefcase.correlation import briefcase_workflow, get_current_workflow

client = Mock()

with briefcase_workflow("support_pipeline", client) as workflow:
    assert get_current_workflow() is workflow

assert get_current_workflow() is None

Propagating context across process boundaries

When an agent lives in another service, carry the trace context with the request so the downstream decision joins the same workflow trace. Inject into outbound headers on the producer side; extract from inbound headers on the consumer side.

from briefcase.correlation import (
    TraceContextCarrier,
    inject_trace_context,
    extract_trace_context,
)

# Producer service: inject the active trace context into outbound headers.
headers = inject_trace_context({})
# ... send headers to the downstream agent ...

# Consumer service: restore the trace context from inbound headers.
context = extract_trace_context(headers)

# TraceContextCarrier offers the same inject/extract pair as a class.
carrier = TraceContextCarrier()
outbound = carrier.inject()
TraceContextCarrier.extract(outbound)

inject_trace_context() reads the active span context; run under an OpenTelemetry tracer for the headers to carry traceparent.

Events

Events are typed signals you emit as decisions happen, so you can react in real time instead of polling the log. The emit functions are coroutines — await them inside an async context.

Emitting an event

Construct a BriefcaseEvent and await emit(...). The idempotency_key lets downstream consumers deduplicate retries.

import asyncio
from briefcase.events import (
    BriefcaseEvent,
    emit,
    emit_low_confidence,
    emit_drift_detected,
)


class Decision:
    def __init__(self, decision_id: str):
        self.decision_id = decision_id


async def main() -> None:
    decision = Decision("dec-91f2")

    event = BriefcaseEvent(
        event_type="decision.recorded",
        decision_id=decision.decision_id,
        payload={"category": "billing", "confidence": 0.62},
        idempotency_key="dec-91f2:recorded",
    )
    await emit(event)

    # The classifier came back unsure — fires only when confidence is below threshold.
    await emit_low_confidence(decision, confidence=0.62, threshold=0.75)

    # A monitored decision drifted — fires when repeated runs disagree.
    await emit_drift_detected(decision, details={"agreement_rate": 0.4})


asyncio.run(main())

Set webhook_url, webhook_secret, events, or event_bus on setup() to route emitted events to a destination.

Event functions

Function	Fires for
`emit(event)`	Any `BriefcaseEvent` you construct
`emit_low_confidence(decision, confidence, threshold)`	A decision below a confidence threshold
`emit_drift_detected(decision, details=None)`	Disagreement across repeated runs

emit_low_confidence pairs naturally with the confidence score on a classify_ticket decision; emit_drift_detected is the live counterpart to drift detection.

Kafka and webhook transports

KafkaPublisher serializes each event as JSON and keys it by idempotency_key. WebhookEmitter emits CloudEvents headers and an X-Briefcase-Signature HMAC-SHA256 digest.

from briefcase.events.kafka import KafkaPublisher
from briefcase.events.webhook import WebhookEmitter

kafka = KafkaPublisher(["kafka:9092"], "briefcase-events")
webhook = WebhookEmitter("https://events.example/briefcase", secret="shared")

Webhooks reject non-loopback plain HTTP by default and refuse redirects. Kafka requires the kafka extra; webhooks use the standard library.

Key symbols

briefcase_workflow(name, client) — context manager yielding the workflow context.
workflow.workflow_id / workflow.register_agent(agent_id, agent_type) — the shared id and agent registration.
get_current_workflow() — the active workflow, or None outside a block.
inject_trace_context / extract_trace_context / TraceContextCarrier — carry context across boundaries.
BriefcaseEvent, emit, emit_low_confidence, emit_drift_detected — the event surface (coroutines).

Limits

Correlation writes to traces, not to your records. A @capture record carries decision_id, decision_type, function_name, inputs, outputs, and timing. It has no workflow_id. Opening a workflow does not change that. Attach it yourself where you want it queryable:

from unittest.mock import Mock
import briefcase
from briefcase.correlation import briefcase_workflow, get_current_workflow

mem = briefcase.observe("memory")


@briefcase.capture(async_capture=False)
def classify_ticket(text: str, workflow_id: str) -> str:
    return "account_access"


with briefcase_workflow("support_pipeline", Mock()) as workflow:
    classify_ticket("reset my password", get_current_workflow().workflow_id)

print(mem.records[-1]["inputs"]["args"])   # the workflow id is in the record

Without OpenTelemetry there is no chain. The root span, the agent_started events from register_agent, and the agent-chain attribute all go to a tracer. With opentelemetry absent or unconfigured, the context manager still yields a workflow with an id and still counts agents, and emits nothing. See OpenTelemetry for wiring one up.

The active workflow is thread-local. get_current_workflow() returns None in a thread you spawned inside the block, so a worker pool sees no workflow unless you pass the id across yourself.

register_agent is bookkeeping. It increments a count, appends to the chain string, and adds a span event. It writes nothing to storage and never calls the client you passed in; that argument is held for your agents to use.

extract_trace_context(headers) returns a contextvars.Token, not a context object. It restores the context as a side effect; keep the token if you want to reset it, and do not treat the return value as the context itself.

inject_trace_context({}) returns the mapping unchanged when no span is active. An empty dict back means there was nothing to propagate, not that propagation failed.

API reference

briefcase.correlation and briefcase.events have the full signatures.

Best practices

Use descriptive workflow names — support_pipeline beats wf-7 when you review later.
Register every agent — unregistered agents leave gaps in the chain.
Propagate context across boundaries — otherwise a remote agent starts its own trace.
Emit events at decision points — low confidence and drift are the signals worth acting on first.

Where this fits

Correlation and events are part of operating a governed system: they keep multi-agent runs accountable and let you react as decisions happen.

OpenTelemetry The trace timeline that workflow context propagates across.

Drift Detection The monitoring behind emit_drift_detected — verify decisions hold over time.