Skip to content

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 
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
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 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
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 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
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 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
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 
64
65
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)

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 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
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 
365
366
367
368
369
370
371
372
373
374
375
376
377
378
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 
380
381
382
383
384
385
386
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 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
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 
156
157
158
159
160
161
162
163
164
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()