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

When you’d reach for this

Your support-triage pipeline is three agents: one retrieves prior tickets, one runs classify_ticket, one drafts a reply. When a customer escalates, you need to see all three decisions as one chain — not three unrelated records. Workflow correlation gives them a shared workflow_id, and the event surface lets an on-call alert fire the moment a classification comes back low-confidence.

Why correlation matters

Before 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.

After correlation

All three decisions share one workflow_id. 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 — accountability across agent boundaries, not just within one agent.

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]

Note

Neither the correlation nor events extra installs third-party dependencies.

---

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.

1. Open a workflow — briefcase_workflow("support_pipeline", client) gives every agent inside it one shared workflow_id.

2. Retrieve — the retrieval agent registers under that workflow.

3. Classify — classify_ticket’s agent registers under the same workflow.

4. Decide — the drafting agent registers under the same workflow.

5. 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.

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).

Best practices

1. Use descriptive workflow names — support_pipeline beats wf-7 when you review later.

2. Register every agent — unregistered agents leave gaps in the chain.

3. Propagate context across boundaries — otherwise a remote agent starts its own trace.

4. 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.