Extension engines & integrations

Framework extensions that live under lazybridge.ext.* — pip install lazybridge ships them by default (except OTelExporter, which requires the [otel] extra).

Connectors moved (0.8). The MCP connector and the external tool gateway are no longer lazybridge.ext.* — they moved to the LazyTools package (lazytools.connectors.{mcp,gateway}, pip install lazytoolkit). The old lazybridge.ext.{mcp,gateway} deprecation shims were removed in 0.9 — import from lazytools instead.

For narrative usage see the corresponding guides: HumanEngine, SupervisorEngine, MCP, Evals, OpenTelemetry, Visualizer.

Human-in-the-loop

lazybridge.ext.hil.HumanEngine

HumanEngine(*, timeout: float | None = None, ui: Literal['terminal', 'web'] | _UIProtocol = 'terminal', default: str | None = None)

Presents the task to a human and returns their response as an Envelope.

With output=PydanticModel, terminal prompts each field; web renders a form. Emits the same 8 event types as LLMEngine for transparent observability.

Source code in lazybridge/ext/hil/human.py

def __init__(
    self,
    *,
    timeout: float | None = None,
    ui: Literal["terminal", "web"] | _UIProtocol = "terminal",
    default: str | None = None,
) -> None:
    self.timeout = timeout
    self.default = default
    if isinstance(ui, str):
        if ui == "terminal":
            self._ui: _UIProtocol = _TerminalUI(timeout=timeout, default=default)
        elif ui == "web":
            self._ui = _WebUI(timeout=timeout, default=default)
        else:
            raise ValueError(f"Unknown UI type: {ui!r}")
    else:
        self._ui = ui

lazybridge.ext.hil.SupervisorEngine

SupervisorEngine(*, tools: list[Tool | Callable | Any] | None = None, agents: list[Any] | None = None, store: Store | None = None, input_fn: Callable[[str], str] | None = None, ainput_fn: Callable[[str], Awaitable[str]] | None = None, timeout: float | None = None, default: str | None = None)

Human-in-the-loop engine with tool-calling and agent retry.

Source code in lazybridge/ext/hil/supervisor.py

def __init__(
    self,
    *,
    tools: list[Tool | Callable | Any] | None = None,
    agents: list[Any] | None = None,
    store: Store | None = None,
    input_fn: Callable[[str], str] | None = None,
    ainput_fn: Callable[[str], Awaitable[str]] | None = None,
    timeout: float | None = None,
    default: str | None = None,
) -> None:
    # Tool-is-Tool: accept plain functions and Agents too, not just Tool
    # instances.  Matches the contract of ``Agent(tools=[...])`` so the
    # same tools list can be handed to either surface.
    from lazybridge.tools import _wrap_tool

    wrapped = [_wrap_tool(t) for t in (tools or [])]
    self._tools = {t.name: t for t in wrapped}
    self._agents = {getattr(a, "name", f"agent-{i}"): a for i, a in enumerate(agents or [])}
    self._store = store
    self._input_fn = input_fn or (lambda prompt: input(prompt))
    self._ainput_fn = ainput_fn
    self.timeout = timeout
    self.default = default

lazybridge.ext.hil.human_agent

human_agent(*, timeout: float | None = None, ui: Literal['terminal', 'web'] | Any = 'terminal', default: str | None = None, **agent_kwargs: Any) -> Agent

Build a human-input :class:Agent (approval gate / form-style HIL).

Symmetric counterpart of Agent.from_<kind>(...) for the :class:HumanEngine. Use this for synchronous human input — a prompt at the terminal or a web form — rather than the full REPL of :func:supervisor_agent.

Engine kwargs (timeout, ui, default) configure the :class:HumanEngine; remaining **agent_kwargs flow to the unified Agent constructor::

from lazybridge.ext.hil import human_agent

human_agent(timeout=60.0, default="approve")("Approve deploy?")

Source code in lazybridge/ext/hil/__init__.py

def human_agent(
    *,
    timeout: float | None = None,
    ui: Literal["terminal", "web"] | Any = "terminal",
    default: str | None = None,
    **agent_kwargs: Any,
) -> Agent:
    """Build a human-input :class:`Agent` (approval gate / form-style HIL).

    Symmetric counterpart of ``Agent.from_<kind>(...)`` for the
    :class:`HumanEngine`.  Use this for **synchronous human input** —
    a prompt at the terminal or a web form — rather than the full REPL
    of :func:`supervisor_agent`.

    Engine kwargs (``timeout``, ``ui``, ``default``) configure the
    :class:`HumanEngine`; remaining ``**agent_kwargs`` flow to the
    unified Agent constructor::

        from lazybridge.ext.hil import human_agent

        human_agent(timeout=60.0, default="approve")("Approve deploy?")
    """
    from lazybridge import Agent

    engine = HumanEngine(timeout=timeout, ui=ui, default=default)
    # 0.7.9 requires explicit name= on non-LLM engines.  Supply the
    # canonical default for the human-input factory; explicit ``name=``
    # in ``agent_kwargs`` wins.
    agent_kwargs.setdefault("name", "human")
    return Agent(engine=engine, **agent_kwargs)

lazybridge.ext.hil.supervisor_agent

supervisor_agent(*, tools: list[Any] | None = None, agents: list[Any] | None = None, store: Any | None = None, input_fn: Callable[[str], str] | None = None, ainput_fn: Callable[[str], Awaitable[str]] | None = None, timeout: float | None = None, default: str | None = None, **agent_kwargs: Any) -> Agent

Build a human-supervised :class:Agent (REPL + tool dispatch + retry).

Symmetric counterpart of Agent.from_<kind>(...) for the :class:SupervisorEngine. Kept on the ext side rather than as Agent.from_supervisor to respect the core/ext import boundary (see docs/guides/core-vs-ext.md).

Engine kwargs (tools, agents, store, input_fn / ainput_fn, timeout, default) configure the :class:SupervisorEngine; remaining **agent_kwargs (memory= / session= / output= / verify= / fallback= / guard= / name= / etc.) flow to the unified Agent constructor::

from lazybridge.ext.hil import supervisor_agent

supervisor_agent(
    tools=[search],
    agents=[researcher],   # human can `retry researcher: <feedback>`
    session=sess,
    name="ops-supervisor",
)("publish a policy brief")

Source code in lazybridge/ext/hil/__init__.py

def supervisor_agent(
    *,
    tools: list[Any] | None = None,
    agents: list[Any] | None = None,
    store: Any | None = None,
    input_fn: Callable[[str], str] | None = None,
    ainput_fn: Callable[[str], Awaitable[str]] | None = None,
    timeout: float | None = None,
    default: str | None = None,
    **agent_kwargs: Any,
) -> Agent:
    """Build a human-supervised :class:`Agent` (REPL + tool dispatch + retry).

    Symmetric counterpart of ``Agent.from_<kind>(...)`` for the
    :class:`SupervisorEngine`.  Kept on the ext side rather than as
    ``Agent.from_supervisor`` to respect the core/ext import boundary
    (see ``docs/guides/core-vs-ext.md``).

    Engine kwargs (``tools``, ``agents``, ``store``, ``input_fn`` /
    ``ainput_fn``, ``timeout``, ``default``) configure the
    :class:`SupervisorEngine`; remaining ``**agent_kwargs`` (``memory=`` /
    ``session=`` / ``output=`` / ``verify=`` / ``fallback=`` / ``guard=`` /
    ``name=`` / etc.) flow to the unified Agent constructor::

        from lazybridge.ext.hil import supervisor_agent

        supervisor_agent(
            tools=[search],
            agents=[researcher],   # human can `retry researcher: <feedback>`
            session=sess,
            name="ops-supervisor",
        )("publish a policy brief")
    """
    # Local import — ``Agent`` lives in core, but core never imports
    # from ext, only the reverse, so this is the architecturally
    # correct direction.
    from lazybridge import Agent

    engine = SupervisorEngine(
        tools=tools,
        agents=agents,
        store=store,
        input_fn=input_fn,
        ainput_fn=ainput_fn,
        timeout=timeout,
        default=default,
    )
    # 0.7.9 requires explicit name= on non-LLM engines.  ``supervisor_agent``
    # is the one-line ergonomic factory — give it a sensible default
    # (``"supervisor"``) when the caller didn't pass one.  An explicit
    # ``name=`` in ``agent_kwargs`` still wins.
    agent_kwargs.setdefault("name", "supervisor")
    return Agent(engine=engine, **agent_kwargs)

MCP integration

Moved to lazytools.connectors.mcp — see the MCP guide and the LazyTools overview. Install with pip install lazytoolkit[mcp].

Evaluation framework

lazybridge.ext.evals.EvalSuite

EvalSuite(*cases: EvalCase)

Run a set of EvalCases against any agent callable.

Source code in lazybridge/ext/evals/__init__.py

def __init__(self, *cases: EvalCase) -> None:
    self.cases = list(cases)

lazybridge.ext.evals.EvalCase dataclass

EvalCase(input: str, check: Callable[..., bool], expected: Any = None, description: str = '')

lazybridge.ext.evals.EvalReport dataclass

EvalReport(results: list[EvalResult] = list())

lazybridge.ext.evals.EvalResult dataclass

EvalResult(case: EvalCase, output: str, passed: bool, error: str | None = None)

Assertion helpers

Ready-made assertion callables for EvalCase (compose your own too):

lazybridge.ext.evals.exact_match

exact_match(expected: str) -> Callable[[str, str], bool]

Source code in lazybridge/ext/evals/__init__.py

def exact_match(expected: str) -> Callable[[str, str], bool]:
    return lambda output, exp: output.strip() == exp.strip()

lazybridge.ext.evals.contains

contains(substring: str) -> Callable[[str], bool]

Source code in lazybridge/ext/evals/__init__.py

def contains(substring: str) -> Callable[[str], bool]:
    return lambda output: substring.lower() in output.lower()

lazybridge.ext.evals.not_contains

not_contains(substring: str) -> Callable[[str], bool]

Source code in lazybridge/ext/evals/__init__.py

def not_contains(substring: str) -> Callable[[str], bool]:
    return lambda output: substring.lower() not in output.lower()

lazybridge.ext.evals.min_length

min_length(n: int) -> Callable[[str], bool]

Source code in lazybridge/ext/evals/__init__.py

def min_length(n: int) -> Callable[[str], bool]:
    return lambda output: len(output) >= n

lazybridge.ext.evals.max_length

max_length(n: int) -> Callable[[str], bool]

Source code in lazybridge/ext/evals/__init__.py

def max_length(n: int) -> Callable[[str], bool]:
    return lambda output: len(output) <= n

lazybridge.ext.evals.llm_judge

llm_judge(agent: Any, criteria: str) -> Callable[[str], bool]

Returns a judge function using an agent to evaluate output.

Verdict recognition uses the same robust normaliser as :func:lazybridge._verify.verify_with_retry (W1.1): the judge may return any of approved / accept / allow / pass / ok / yes / good / valid (synonyms, case-insensitive, prefix-anchored) to approve. Explicit reject prefixes (rejected / deny / block / fail / no / bad / invalid) and any unrecognised verdict are treated as rejection (fail-safe) — so a judge that fails to produce a clean verdict never accidentally passes a bad output.

Source code in lazybridge/ext/evals/__init__.py

def llm_judge(agent: Any, criteria: str) -> Callable[[str], bool]:
    """Returns a judge function using an agent to evaluate output.

    Verdict recognition uses the same robust normaliser as
    :func:`lazybridge._verify.verify_with_retry` (W1.1): the judge may
    return any of ``approved`` / ``accept`` / ``allow`` / ``pass`` /
    ``ok`` / ``yes`` / ``good`` / ``valid`` (synonyms,
    case-insensitive, prefix-anchored) to approve.  Explicit reject
    prefixes (``rejected`` / ``deny`` / ``block`` / ``fail`` / ``no``
    / ``bad`` / ``invalid``) and any unrecognised verdict are treated
    as rejection (fail-safe) — so a judge that fails to produce a
    clean verdict never accidentally passes a bad output.
    """
    # Import lazily to avoid a top-of-module dependency on a private
    # core helper for users who don't use llm_judge.
    from lazybridge._verify import _is_approved

    def judge(output: str) -> bool:
        verdict = agent(f"Criteria: {criteria}\nOutput to judge: {output}\nVerdict (approved/rejected):").text()
        return _is_approved(verdict)

    return judge

Planners

Multi-step planning agents (lazybridge.ext.planners). orchestrator_agent / blackboard_orchestrator_agent are the canonical factories; make_planner / make_blackboard_planner are backward-compat aliases for the same callables. See the Planners guide.

lazybridge.ext.planners.orchestrator_agent module-attribute

orchestrator_agent = make_planner

lazybridge.ext.planners.blackboard_orchestrator_agent module-attribute

blackboard_orchestrator_agent = make_blackboard_planner

lazybridge.ext.planners.make_plan_builder_tools

make_plan_builder_tools(registry: dict[str, Agent], *, max_plans: int = 50) -> list[Tool]

Five builder tools that share state via closure.

Returns [create_plan, add_step, inspect_plan, run_plan, discard_plan].

The state is per-factory-instance — call make_plan_builder_tools fresh for each planner agent (or each session) if you want isolated blackboards. run_plan and discard_plan consume the plan from the dict, so memory stays bounded as long as the planner finishes its plans. max_plans is a hard cap on concurrent in-progress plans (oldest-evicted on overflow) so a misbehaving planner can't leak memory.

Source code in lazybridge/ext/planners/builder.py

def make_plan_builder_tools(
    registry: dict[str, Agent],
    *,
    max_plans: int = 50,
) -> list[Tool]:
    """Five builder tools that share state via closure.

    Returns ``[create_plan, add_step, inspect_plan, run_plan, discard_plan]``.

    The state is per-factory-instance — call ``make_plan_builder_tools``
    fresh for each planner agent (or each session) if you want isolated
    blackboards. ``run_plan`` and ``discard_plan`` consume the plan from
    the dict, so memory stays bounded as long as the planner finishes its
    plans. ``max_plans`` is a hard cap on concurrent in-progress plans
    (oldest-evicted on overflow) so a misbehaving planner can't leak memory.
    """
    if not registry:
        raise ValueError("plan tool registry must contain at least one agent")

    plans: dict[str, _PlanInProgress] = {}

    def _evict_if_full() -> None:
        if len(plans) >= max_plans:
            # Drop the oldest in-progress plan.
            oldest = min(plans.values(), key=lambda p: p.created_at)
            plans.pop(oldest.plan_id, None)

    # --- create_plan -----------------------------------------------------
    def create_plan(reasoning: str) -> str:
        """Start a new empty plan. Returns the plan_id to use in subsequent calls.

        Args:
            reasoning: Why this plan; which sub-agents and why; simplest
                shape that fits. Required — empty / boilerplate defeats
                the point of thinking first.
        """
        if not reasoning or not reasoning.strip():
            return (
                "REJECTED: reasoning is required and must be non-empty. "
                "Briefly state why this plan shape fits the task."
            )
        _evict_if_full()
        pid = uuid.uuid4().hex[:8]
        plans[pid] = _PlanInProgress(plan_id=pid, reasoning=reasoning.strip())
        return (
            f"plan_id={pid} (empty; add steps with add_step, then run_plan). "
            f"Available sub-agents: {sorted(registry)!r}."
        )

    # --- add_step --------------------------------------------------------
    def add_step(
        plan_id: str,
        name: str,
        agent: str,
        task_kind: Literal["literal", "from_prev", "from_step", "from_parallel", "from_parallel_all"] = "from_prev",
        task_text: str | None = None,
        task_step: str | None = None,
        context_kind: Literal["from_step", "from_parallel"] | None = None,
        context_step: str | None = None,
        parallel: bool = False,
    ) -> str:
        """Append one step to a plan; validated immediately.

        On rejection, the plan is unchanged — fix the args and call again.

        Args:
            plan_id: From a prior ``create_plan``.
            name: Unique snake_case identifier within this plan.
            agent: Sub-agent name (must exist in the registry).
            task_kind: ``literal`` (use ``task_text``) / ``from_prev``
                (default; previous step's output) / ``from_step``
                (named earlier step's output) / ``from_parallel``
                (alias of ``from_step``, naming is for readability) /
                ``from_parallel_all`` (aggregate the WHOLE parallel band
                starting at ``task_step`` into one labelled-text join;
                ``task_step`` must be the FIRST ``parallel=true`` member).
            task_text: Required when ``task_kind="literal"``.
            task_step: Required when ``task_kind`` is ``from_step``,
                ``from_parallel``, or ``from_parallel_all``; must name an
                earlier step.
            context_kind: Optional secondary input pulled into the step's
                context. Useful to combine TWO parallel branches.
            context_step: Required when ``context_kind`` is set.
            parallel: ``true`` to run concurrently with adjacent
                ``parallel=true`` siblings.
        """
        if plan_id not in plans:
            return f"REJECTED: unknown plan_id {plan_id!r}."
        pip = plans[plan_id]
        err = _validate_step_addition(
            pip,
            name,
            agent,
            task_kind,
            task_text,
            task_step,
            context_kind,
            context_step,
            registry,
        )
        if err:
            return f"REJECTED: {err}"
        pip.steps.append(
            StepSpec(
                name=name,
                agent=agent,
                task_kind=task_kind,
                task_text=task_text,
                task_step=task_step,
                context_kind=context_kind,
                context_step=context_step,
                parallel=parallel,
            )
        )
        return f"ok ({len(pip.steps)} step(s) in plan {plan_id})"

    # --- inspect_plan ----------------------------------------------------
    def inspect_plan(plan_id: str) -> str:
        """Show the plan's current shape — useful between additions."""
        if plan_id not in plans:
            return f"REJECTED: unknown plan_id {plan_id!r}."
        return _format_progress(plans[plan_id])

    # --- run_plan --------------------------------------------------------
    async def run_plan(plan_id: str, task: str) -> str:
        """Materialise and run the plan; returns the final step's text.

        Consumes the plan (it's removed from the in-progress dict). To
        run again, build a new plan.
        """
        if plan_id not in plans:
            return f"REJECTED: unknown plan_id {plan_id!r}."
        pip = plans.pop(plan_id)
        if not pip.steps:
            return f"REJECTED: plan {plan_id} has no steps. Add at least one before running."
        spec = PlanSpec(reasoning=pip.reasoning, task=task, steps=pip.steps)
        try:
            plan = _materialize(spec, registry)
        except _PlanToolError as e:
            return f"PLAN_REJECTED: {e}"
        try:
            # 0.7.9 requires explicit name= on non-LLM engines; this throw-away
            # runner only exists to materialise + execute the plan once, so any
            # stable identifier suffices.
            runner = Agent(engine=plan, name=f"_planner_runner_{plan_id}")  # PlanCompiler defense-in-depth.
        except PlanCompileError as e:
            return _format_compile_error(e, registry)
        try:
            env = await runner.run(spec.task)
        except Exception as e:
            return f"PLAN_RUNTIME_ERROR: {type(e).__name__}: {e}"
        if env.error:
            return f"PLAN_RUNTIME_ERROR: {env.error.message}"
        return env.text()

    # --- discard_plan ----------------------------------------------------
    def discard_plan(plan_id: str) -> str:
        """Drop an in-progress plan without running it."""
        if plan_id not in plans:
            return f"REJECTED: unknown plan_id {plan_id!r}."
        plans.pop(plan_id)
        return f"ok (plan {plan_id} discarded)"

    # Customise descriptions so the LLM sees the registry inline.
    agents_summary = "Available sub-agents:\n" + "\n".join(
        f"- {n}: {(a.description or '').strip() or 'no description'}" for n, a in registry.items()
    )

    add_step.__doc__ = (add_step.__doc__ or "") + "\n\n" + agents_summary
    create_plan.__doc__ = (create_plan.__doc__ or "") + "\n\n" + agents_summary

    return [
        Tool(create_plan, mode="signature"),
        Tool(add_step, mode="signature"),
        Tool(inspect_plan, mode="signature"),
        Tool(run_plan, mode="signature"),
        Tool(discard_plan, mode="signature"),
    ]

lazybridge.ext.planners.make_execute_plan_tool

make_execute_plan_tool(registry: dict[str, Agent], **_unused: Any) -> list[Tool]

Deprecated alias for :func:make_plan_builder_tools. Returns the list of builder tools rather than a single Tool.

Source code in lazybridge/ext/planners/builder.py

def make_execute_plan_tool(registry: dict[str, Agent], **_unused: Any) -> list[Tool]:
    """Deprecated alias for :func:`make_plan_builder_tools`. Returns the
    list of builder tools rather than a single Tool."""
    return make_plan_builder_tools(registry)

lazybridge.ext.planners.PlanSpec

Bases: BaseModel

The argument shape of execute_plan.

lazybridge.ext.planners.StepSpec

Bases: BaseModel

One node in the plan DAG.

Tiered approval gate

Declarative, per-agent ApprovalGate policy for coding engines (lazybridge.ext.approval) — an ordered Rule table assigns each tool one of four tiers (allow/session/ask/deny); unmatched calls are denied by default. TieredGate implements ApprovalGate directly (no wrapper). Promoted from the approval-lab prototype, with session grants scoped to (provider, kind, name, cwd, policy fingerprint) and every decision recorded in a structured AuditRecord.

lazybridge.ext.approval.TieredGate dataclass

TieredGate(channel: Channel, rules: tuple[Rule, ...], on_record: Any = None, _session_grants: set[tuple[str, str, str, str, str]] = set(), log: list[AuditRecord] = list())

ApprovalGate implementation driven by an ordered rule table.

Implements :class:~lazybridge.engines.coding.ApprovalGate directly (structural typing — async def __call__(self, request) -> ApprovalDecision is the whole contract): no bridge class re-declares the signature.

lazybridge.ext.approval.Rule dataclass

Rule(tier: Tier, name_pattern: str, arg_pattern: str | None = None)

fingerprint

fingerprint() -> str

Short stable hash of this rule's shape, used to scope session grants.

Two rules that look identical fingerprint identically even if they are different Rule instances (e.g. rebuilt on every process start) — the fingerprint is over content, not identity.

Source code in lazybridge/ext/approval/tiered.py

def fingerprint(self) -> str:
    """Short stable hash of this rule's shape, used to scope session grants.

    Two rules that look identical fingerprint identically even if they
    are different ``Rule`` instances (e.g. rebuilt on every process
    start) — the fingerprint is over content, not identity.
    """
    payload = f"{self.tier}|{self.name_pattern}|{self.arg_pattern or ''}"
    return hashlib.sha256(payload.encode()).hexdigest()[:16]

lazybridge.ext.approval.AuditRecord dataclass

AuditRecord(timestamp: float, provider: str, kind: str, tool_name: str, cwd: str | None, tier: str, action: str, message: str, arguments_hash: str, arguments_preview: str, responder: str | None = None)

One structured audit entry — every decision the gate makes, allowed or not.

arguments_hash/arguments_preview stand in for the raw arguments so the audit log stays safe to retain and to display even when a call carries large payloads or sensitive-looking values (file contents, tokens pasted into a command). The hash is over the exact JSON the rule matched against, so two records with equal hashes really did see equal arguments.

lazybridge.ext.approval.Channel

Bases: Protocol

A human-facing yes/no prompt. Implementations: terminal, chat, queue.

Only the ask coroutine is part of the structural contract. A channel MAY additionally expose a name: str attribute identifying itself (e.g. "terminal", "telegram") — :class:TieredGate reads it via getattr for the audit trail and falls back to the class name when absent, so plain implementations need not declare it.

lazybridge.ext.approval.TerminalChannel

Ask on stdin. y/yes/si/s/ok approve; anything else denies.

Generic and dependency-free — useful for local development and for exercising :class:TieredGate in tests without a real chat backend. Production agents that run unattended (scheduled jobs, always-on processes) need a different channel; this one blocks on a human being at the terminal.

OpenTelemetry exporter

lazybridge.ext.otel.OTelExporter

OTelExporter(*, endpoint: str | None = None, exporter: Any | None = None, batch: bool = True)

Export events as OpenTelemetry spans (requires opentelemetry-sdk).

Emits gen_ai.* attributes per the OpenTelemetry Semantic Conventions for GenAI, with proper parent-child span hierarchy.

Install: pip install lazybridge[otel]

Thread-safe: Session.emit can fan events out to this exporter from multiple worker threads, so the in-flight span registry is guarded by a lock. Call :meth:close to flush any spans that are still open (e.g. when a run is cancelled before agent_finish).

The exporter sets each span as the current OTel context span while it is open, so nested agents (Agent-as-tool) automatically inherit the outer tool span as their parent without any explicit correlation id — OTel's contextvars-based propagation does the work.

Source code in lazybridge/ext/otel/exporter.py

def __init__(
    self,
    *,
    endpoint: str | None = None,
    exporter: Any | None = None,
    batch: bool = True,
) -> None:
    try:
        from opentelemetry.sdk.trace import TracerProvider
        from opentelemetry.sdk.trace.export import BatchSpanProcessor, SimpleSpanProcessor

        provider = TracerProvider()
        _proc = BatchSpanProcessor if batch else SimpleSpanProcessor
        if exporter:
            provider.add_span_processor(_proc(exporter))
        elif endpoint:
            from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter

            provider.add_span_processor(_proc(OTLPSpanExporter(endpoint=endpoint)))
        self._provider = provider
        self._tracer = provider.get_tracer("lazybridge")
    except ImportError:
        raise ImportError("Install opentelemetry-sdk: pip install lazybridge[otel]") from None

    # Outer key: run_id.  Inner key: ``"agent"``, ``"model"``, or
    # ``f"tool:{tool_use_id_or_name}"``.  Guarded by ``self._lock``.
    self._spans: dict[str, dict[str, _SpanEntry]] = {}
    self._lock = threading.Lock()

close

close() -> None

Flush any spans still open — e.g. after a cancelled run.

Idempotent. Without this, a run that crashes before emitting agent_finish leaves its spans stuck for the life of the process, and the OTel contextvars stay attached to nothing.

Source code in lazybridge/ext/otel/exporter.py

def close(self) -> None:
    """Flush any spans still open — e.g. after a cancelled run.

    Idempotent.  Without this, a run that crashes before emitting
    ``agent_finish`` leaves its spans stuck for the life of the
    process, and the OTel contextvars stay attached to nothing.
    """
    with self._lock:
        run_ids = list(self._spans.keys())
    for run_id in run_ids:
        with self._lock:
            keys = list(self._spans.get(run_id, {}).keys())
        for key in keys:
            self._end_span(run_id, key, error="exporter closed")

flush

flush(timeout_millis: int = 30000) -> None

Drain pending spans from the BatchSpanProcessor.

No-op when batch=False (SimpleSpanProcessor flushes synchronously). Call before process exit to ensure all spans reach the collector.

Source code in lazybridge/ext/otel/exporter.py

def flush(self, timeout_millis: int = 30_000) -> None:
    """Drain pending spans from the BatchSpanProcessor.

    No-op when ``batch=False`` (SimpleSpanProcessor flushes synchronously).
    Call before process exit to ensure all spans reach the collector.
    """
    self._provider.force_flush(timeout_millis=timeout_millis)

Visualizer

lazybridge.ext.viz.Visualizer

Visualizer(session: Session, *, store: Store | None = None, host: str = '127.0.0.1', port: int = 0, auto_open: bool = True)

Live or replay visualizer.

Use as a context manager so the HTTP server is shut down cleanly when the with-block exits. The browser stays open across that boundary; the user is expected to close the tab themselves.

Source code in lazybridge/ext/viz/visualizer.py

def __init__(
    self,
    session: Session,
    *,
    store: Store | None = None,
    host: str = "127.0.0.1",
    port: int = 0,
    auto_open: bool = True,
) -> None:
    self._session = session
    self._store = store
    self._hub = EventHub()
    self._exporter = HubExporter(self._hub)
    self._mode = "live"
    self._replay: ReplayController | None = None

    # Wire the hub into the live session
    session.add_exporter(self._exporter)

    self._server = VizServer(
        self._hub,
        graph_provider=self._graph_payload,
        store_provider=self._store_payload,
        meta_provider=self._meta_payload,
        host=host,
        port=port,
    )
    self._auto_open = auto_open
    self._opened = False

open

open() -> None

Block the caller until Ctrl+C, useful for replay scripts.

Source code in lazybridge/ext/viz/visualizer.py

def open(self) -> None:
    """Block the caller until Ctrl+C, useful for replay scripts."""
    self.start()
    print(f"[viz] open → {self.url}")
    try:
        while True:
            time.sleep(3600)
    except KeyboardInterrupt:
        self.stop()