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.

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