Agent + Envelope

The universal wrapper and the typed result object every run produces. For narrative usage see Guides → Basic → Agent and Guides → Basic → Envelope. If you're migrating from a 0.7-era surface (Agent.from_* factories, config dataclasses, _ParallelAgent), see the 0.7 → 0.7.9 migration guide.

lazybridge.Agent

Agent(engine: str | Any | None = None, tools: list[Tool | Callable | Agent] | None = None, output: type = str, memory: Any | None = None, store: Any | None = None, sources: list[Any] | None = None, guard: Any | None = None, verify: Agent | Callable[[str], Any] | None = None, max_verify: int = 3, name: str | None = None, description: str | None = None, session: Any | None = None, verbose: bool = False, model: str | None = None, native_tools: list[Any] | None = None, allow_dangerous_native_tools: bool = False, output_validator: Callable[[Any], Any] | None = None, max_output_retries: int = 2, timeout: float | None = None, max_retries: int = 3, retry_delay: float = 1.0, fallback: Agent | None = None, cache: bool | Any = False)

Universal agent — Agent(engine, tools, state).

Every Agent has the same shape, regardless of what it does:

• engine — the brain: decides what happens (LLM, Plan, Human, …)
• tools — the capabilities: what the agent can invoke
• state — memory, session, guard, verify, output

Canonical composition — give each sub-agent an explicit name= and pass it directly in tools=[...]::

from lazybridge import Agent, LLMEngine, Plan, Step, tool, from_prev, from_step

search = tool(search_web, name="search", description="Search the web.")

researcher = Agent(
    name="research",
    engine=LLMEngine("claude-haiku-4-5", system="You are a research expert."),
    tools=[search],
)
writer = Agent(
    name="write",
    engine=LLMEngine("gpt-5.4-mini", system="You are a concise writer."),
)

# Deterministic orchestrator — Plan engine
pipeline = Agent(
    name="pipeline",
    engine=Plan(
        Step("research"),
        Step("write", task=from_prev, context=from_step("research")),
    ),
    tools=[researcher, writer],   # agents passed directly
    session=sess,
)

# Dynamic orchestrator — LLM engine
orchestrator = Agent(
    name="orchestrator",
    engine=LLMEngine("claude-opus-4-8"),
    tools=[researcher, writer],
    session=sess,
)

The engine is the only thing that changes. Everything else — tools, memory, session, guard, output — is the same surface on every Agent.

String shortcut — Agent("claude-opus-4-8") is sugar for Agent(engine=LLMEngine("claude-opus-4-8")). Use the explicit form when you need to configure the engine (system=, max_turns=, thinking=, etc.).

The name chain — Agent(name=...) is the authoritative key that connects every part of the system::

Agent(name="research")      →  tool map key when passed in tools=[researcher]
Step("research")            →  looks up "research" in tool map ✓
from_step("research")       →  reads output of step "research" (in-Plan) ✓
from_agent("research")      →  reads stored output of "research" (cross-run) ✓
from_memory("research")     →  reads live memory of "research" ✓

Advanced alias / backward compat — .as_tool("alias") remains available when you need a different name than the agent's own::

tools=[researcher.as_tool("deep_research")]

Factory methods that build real structure (not pure aliases) live on the class:

• Agent.chain(a, b) — sequential: builds a Plan of one Step per agent.
• Agent.parallel(*agents) — scripted fan-out: returns a ParallelAgent whose __call__ yields one Envelope (labelled-text join across every branch, with transitive cost rollup). For typed per-branch list[Envelope] access call parallel.run_branches(task) (async).
• Agent.from_provider(provider, tier="medium") — resolves a tier alias (cheap / medium / top / …) to that provider's current model.

Extension engines live in :mod:lazybridge.ext to respect the core/ext import boundary::

from lazybridge.ext.hil import HumanEngine, SupervisorEngine
Agent(engine=HumanEngine(timeout=60), tools=[approve])
Agent(engine=SupervisorEngine(tools=[...]))

Source code in lazybridge/agent.py

def __init__(
    self,
    engine: str | Any | None = None,
    tools: list[Tool | Callable | Agent] | None = None,
    output: type = str,
    memory: Any | None = None,
    store: Any | None = None,
    sources: list[Any] | None = None,
    guard: Any | None = None,
    verify: Agent | Callable[[str], Any] | None = None,
    max_verify: int = 3,
    name: str | None = None,
    description: str | None = None,
    session: Any | None = None,
    verbose: bool = False,
    # Convenience: pin a specific model id via ``model=`` on the
    # auto-LLMEngine path.  Only consumed when ``engine=None`` or
    # ``engine=<model_string>``; passing ``model=`` alongside a
    # pre-built engine raises (see the model-vs-engine check below).
    # For tier-alias model selection use ``Agent.from_provider(
    # "anthropic", tier="top")`` — the bare-provider-name shortcut
    # ``Agent("anthropic", ...)`` was removed in 0.7.9.x because it
    # left the model id ambiguous at request time.
    model: str | None = None,
    # Provider-native server-side tools (WEB_SEARCH, CODE_EXECUTION, …).
    # Accepted directly on Agent as a shortcut for
    # ``Agent(engine=LLMEngine(..., native_tools=[...]))``.  Ignored when
    # ``engine=`` is a non-LLM engine.
    native_tools: list[Any] | None = None,
    # Required opt-in for capabilities with broad access (CODE_EXECUTION,
    # COMPUTER_USE).  Forwarded to LLMEngine when auto-created and used
    # to gate the pre-built engine path so callers can't silently bypass
    # the LLMEngine.__init__ check by passing engine= separately.
    allow_dangerous_native_tools: bool = False,
    # --- Resilience kwargs ---
    # Optional post-parse validator.  Runs on the structured ``payload``
    # after schema validation; may raise ValueError to force a
    # retry-with-feedback loop (up to ``max_output_retries``).
    output_validator: Callable[[Any], Any] | None = None,
    max_output_retries: int = 2,
    # Total deadline (seconds) for ``run()``.  ``None`` disables.
    timeout: float | None = None,
    # Provider retry/backoff — forwarded to LLMEngine when the engine
    # is auto-created from a model string.  Ignored when ``engine=``
    # is supplied explicitly (configure on ``LLMEngine`` directly).
    max_retries: int = 3,
    retry_delay: float = 1.0,
    # Fallback agent tried when the primary engine returns an error.
    # ``Agent("claude-opus-4-7", fallback=Agent("gpt-4o"))``.)
    fallback: Agent | None = None,
    # Prompt caching — when True, marks the static prefix (system
    # prompt + tools) as cacheable so providers that support it
    # (Anthropic today; OpenAI/DeepSeek auto-cache; Google uses a
    # different API) serve cache hits at ~10% of input token cost.
    # Pass a ``CacheConfig(ttl="1h")`` instance for the longer
    # Anthropic TTL.  Forwarded to LLMEngine when the engine is
    # auto-created.  Ignored when ``engine=`` is supplied explicitly
    # (configure ``LLMEngine(cache=...)`` directly).
    cache: bool | Any = False,
) -> None:
    # ``name`` is "explicit" when the caller supplied a real string
    # value (not None / blank).  Used downstream to require a name
    # when the agent is later passed in ``tools=[...]``.
    _name_explicit_flag: bool = name is not None and str(name).strip() != ""
    from lazybridge.engines.llm import LLMEngine

    # Phase-3 Block H, T6 — ``model=`` is only meaningful on the LLM-engine
    # construction path (engine is None or a model-string).  Passing both
    # ``model=`` and a non-string ``engine=`` was silently dropped pre-0.8;
    # 0.7.9 raises so the typo / misconfiguration is visible.
    if model is not None and engine is not None and not isinstance(engine, str):
        raise ValueError(
            f"Agent(model={model!r}, engine={type(engine).__name__}(...)): "
            f"the ``model=`` kwarg is only consumed when ``engine=None`` or "
            f"``engine=<model_string>`` (in which case Agent auto-builds an "
            f"``LLMEngine``).  When you pass a pre-built engine, configure "
            f"the model on that engine itself.\n"
            f"  Fix: drop ``model=`` (engine controls the model), or pass "
            f"the model string directly: ``Agent({model!r}, ...)``."
        )

    # Canonical: Agent(engine=LLMEngine(...)) or Agent(engine=Plan(...))
    # Sugar:     Agent("claude-opus-4-7") → engine is a model string → auto-builds LLMEngine
    #            Agent() → engine is None → defaults to "claude-opus-4-7"
    if engine is None or isinstance(engine, str):
        model_str = model or engine or "claude-opus-4-7"
        self.engine: Any = LLMEngine(
            model_str,
            native_tools=native_tools,
            allow_dangerous_native_tools=allow_dangerous_native_tools,
            max_retries=max_retries,
            retry_delay=retry_delay,
            cache=cache,
        )
    else:
        self.engine = engine
        # Phase-3 Block H, T7 — when the engine isn't an LLM (Plan,
        # SupervisorEngine, HumanEngine, custom), the auto-name fallback
        # to the engine's ``model`` attribute (or to the literal
        # ``"agent"`` placeholder) silently produces ambiguous names that
        # collide once the agent is used as a tool or referenced by a
        # ``Step``.  Require ``name=`` upfront so the failure is at the
        # construction point rather than at first composition.
        if not _name_explicit_flag and not hasattr(self.engine, "model"):
            engine_kind = type(self.engine).__name__
            raise ValueError(
                f"Agent(engine={engine_kind}(...)) requires an explicit ``name=``.\n"
                f"  Engines other than ``LLMEngine`` have no ``.model`` attribute to derive\n"
                f"  a default name from, so the agent would silently get the placeholder\n"
                f"  ``'agent'`` and collide the moment another agent is built or composed.\n"
                f"  Fix: pass ``name=`` (e.g. ``Agent(engine={engine_kind}(...), name='pipeline')``)."
            )

    # If the caller passed native_tools but also supplied a pre-built
    # engine, push the list onto the engine if it has the attribute.
    # This lets ``Agent(engine=LLMEngine("claude"), native_tools=[...])``
    # work the same as ``Agent("claude", native_tools=[...])``.)
    if native_tools and hasattr(self.engine, "native_tools"):
        from lazybridge.core.types import NativeTool

        resolved = [NativeTool(t) if isinstance(t, str) else t for t in native_tools]
        # Run the same dangerous-tools gate that LLMEngine.__init__ would
        # run — prevents bypassing it by passing engine= separately.
        _DANGEROUS = {NativeTool.CODE_EXECUTION, NativeTool.COMPUTER_USE}
        found = [t for t in resolved if t in _DANGEROUS]
        if found and not allow_dangerous_native_tools:
            names = ", ".join(t.value for t in found)
            raise ValueError(
                f"Native tools {names} have broad system access. Pass allow_dangerous_native_tools=True to opt in."
            )
        # Merge without dup — preserve order of existing + append new.
        existing = list(getattr(self.engine, "native_tools", []) or [])
        for t in resolved:
            if t not in existing:
                existing.append(t)
        self.engine.native_tools = existing

    self._tools_raw = list(tools or [])
    # Validate before building the tool map so errors surface early with
    # the agent's current name rather than a wrapped Tool name.
    for _raw in self._tools_raw:
        # Default True so duck-typed agents (MockAgent, custom subclasses)
        # that predate _name_explicit are not rejected.  Only real Agent
        # instances explicitly set this to False when no name= was given.
        if getattr(_raw, "_is_lazy_agent", False) and getattr(_raw, "_name_explicit", True) is False:
            _raw_name = getattr(_raw, "name", repr(_raw))
            raise ValueError(
                f"Agent used as a tool must have an explicit name=...\n"
                f"The agent currently has name={_raw_name!r} "
                f"(derived from the model or left as the default).\n\n"
                f"Set an explicit name:\n"
                f'    Agent(name="research", engine=LLMEngine(...))\n\n'
                f"Or use an alias:\n"
                f'    agent.as_tool("research")\n'
                f'    tool(agent, name="research")'
            )
    self._tool_map: dict[str, Tool] = build_tool_map(self._tools_raw)
    self.output = output
    self.output_validator = output_validator
    self.max_output_retries = max_output_retries
    self.timeout = timeout
    self.memory = memory
    self.store = store
    self.sources = list(sources or [])
    if max_verify < 1:
        raise ValueError(f"max_verify must be >= 1, got {max_verify!r}")
    if max_output_retries < 0:
        raise ValueError(f"max_output_retries must be >= 0, got {max_output_retries!r}")
    self.guard = guard
    self.verify = verify
    self.max_verify = max_verify
    self.fallback = fallback
    if self.fallback is not None:
        seen: set[int] = {id(self)}
        fb: Agent | None = self.fallback
        while fb is not None:
            if id(fb) in seen:
                raise ValueError("fallback= chain contains a cycle. Check your Agent(fallback=...) configuration.")
            seen.add(id(fb))
            fb = getattr(fb, "fallback", None)
    self.name: str = str(name or getattr(self.engine, "model", None) or "agent")
    self.description = description
    #: True when the caller supplied an explicit ``name=``.  False
    #: when the name was derived from the model string or left as
    #: the ``"agent"`` default.
    #: Used by ``build_tool_map`` and the ``tool()`` factory to require
    #: an explicit identity before an Agent is used as a sub-agent tool.
    self._name_explicit: bool = _name_explicit_flag

    #: True when this agent's ``session`` was inherited from a parent
    #: orchestrator (not passed by the user).  Lets a second
    #: orchestrator warn instead of silently misattributing events.
    self._session_inherited: bool = False

    # Private per-agent console exporter when verbose= is set without
    # an explicit Session. Attached when we bind to a session below.
    self._verbose = verbose

    # Bind to session — create an implicit private Session if verbose=
    # is requested without one, so events print to stdout out of the box.
    if session is None and verbose:
        from lazybridge.session import Session

        session = Session(console=True)
    self.session = session

    # Fallback identity for code that drives the engine directly (tests,
    # custom callers).  When the engine runs through this Agent, the
    # per-invocation binding in ``_run_engine``/``_stream_engine`` wins —
    # see ``lazybridge.engines.base.resolve_agent_name`` — so two Agents
    # sharing one engine no longer misattribute events to whichever was
    # constructed last.
    self.engine._agent_name = self.name

    # Register with session graph so it's visible in session.graph.to_json()
    _safe_register_agent(self.session, self)

    # Propagate session to nested Agents passed as tools (they become
    # part of the same observability surface — events from B called
    # via A flow into A's EventLog). Agents that already have a
    # session keep it. This is the fix for the "as_tool" observability
    # paradox: without it, calling B through A's tool loop recorded
    # nothing anywhere.
    if self.session is not None:
        for raw in self._tools_raw:
            if not getattr(raw, "_is_lazy_agent", False):
                continue
            agent_raw = cast("Agent", raw)
            child_session = getattr(agent_raw, "session", None)

            if child_session is None:
                # Propagate parent session down to child and register both
                # the agent node and the parent → child edge.  The flag
                # records that the session was inherited (not set by the
                # user) so a later orchestrator can tell the difference.
                agent_raw.session = self.session
                agent_raw._session_inherited = True
                _safe_register_agent(self.session, agent_raw)
                _safe_register_tool_edge(self.session, self, agent_raw, label=agent_raw.name)
            elif child_session is self.session:
                # Child already shares the same session (canonical pattern:
                # all agents built with session= up front).  Register the
                # edge — it was missing because the old guard checked for
                # ``session is None`` only.
                _safe_register_agent(self.session, agent_raw)
                _safe_register_tool_edge(self.session, self, agent_raw, label=agent_raw.name)
            else:
                # Child belongs to a different session — don't steal it.
                # But if that session was itself *inherited* from another
                # orchestrator (not chosen by the user), the child's
                # events will silently flow to the other session's log.
                # Surface the misattribution instead of hiding it.
                if getattr(agent_raw, "_session_inherited", False):
                    import warnings

                    warnings.warn(
                        f"Sub-agent {agent_raw.name!r} already inherited a session from "
                        f"another orchestrator; its events will flow there, not to "
                        f"{self.name!r}'s session.  Pass session= explicitly on the "
                        f"sub-agent (or build a fresh instance per orchestrator) to "
                        f"choose where its events go.",
                        UserWarning,
                        stacklevel=2,
                    )
        # ``fallback=`` and ``verify=`` Agents inherit the same
        # session + graph-registration the tools list gets, so any
        # events they produce (errors handled by the fallback, judge
        # verdicts from verify) flow into the outer EventLog.  Edge
        # labels distinguish provenance.
        for related, label in (
            (self.fallback, "fallback"),
            (self.verify, "verify"),
        ):
            if (
                related is not None
                and getattr(related, "_is_lazy_agent", False)
                and getattr(related, "session", None) is None
            ):
                # Duck-typed ``_is_lazy_agent`` covers Agent and MockAgent
                # (and any other agent-shaped object); cast for the typed
                # graph helpers since the static type of ``related`` now
                # admits a plain Callable via the widened ``verify=``.
                related_agent = cast("Agent", related)
                related_agent.session = self.session
                _safe_register_agent(self.session, related_agent)
                _safe_register_tool_edge(self.session, self, related_agent, label=label)

    # PlanCompiler runs at construction time
    if hasattr(self.engine, "_validate"):
        self.engine._validate(self._tool_map)

stream async

stream(task: str | Envelope, *, images: list[Any] | None = None, audio: Any | None = None) -> AsyncGenerator[str, None]

Stream LLM tokens across the full tool-calling loop.

Guard enforcement. self.guard is checked via acheck_input before the first token is emitted — a blocked input raises :class:ValueError; a modified input replaces the task sent to the provider. When the stream completes, acheck_output runs on the accumulated text: a block raises :class:ValueError after the tokens have been delivered (streaming is irrevocable) and skips the Store write, so consumers that buffer the stream can still discard it.

Fallback. When the engine fails before the first token is emitted and fallback= is configured, the fallback agent's stream() takes over — same contract as :meth:run, with the primary's failure threaded into the fallback's context. Once tokens have been yielded the error propagates instead (the partial output cannot be retracted).

Timeout. self.timeout is the total deadline for the stream — the same meaning it has in :meth:run. Stall detection between chunks is the engine's job (LLMEngine(stream_idle_timeout=...)).

Not applied in streaming (use :meth:run when you need them): verify= and output= validation — tokens are emitted as they arrive, so there is no pre-delivery result to judge or validate.

Multimodal: pass images= / audio= to attach blocks to the streamed turn — same coercion semantics as :meth:run.

Source code in lazybridge/agent.py

async def stream(
    self,
    task: str | Envelope,
    *,
    images: list[Any] | None = None,
    audio: Any | None = None,
) -> AsyncGenerator[str, None]:
    """Stream LLM tokens across the full tool-calling loop.

    **Guard enforcement.** ``self.guard`` is checked via
    ``acheck_input`` before the first token is emitted — a blocked
    input raises :class:`ValueError`; a modified input replaces the
    task sent to the provider.  When the stream completes,
    ``acheck_output`` runs on the accumulated text: a block raises
    :class:`ValueError` *after* the tokens have been delivered
    (streaming is irrevocable) and skips the Store write, so
    consumers that buffer the stream can still discard it.

    **Fallback.** When the engine fails before the first token is
    emitted and ``fallback=`` is configured, the fallback agent's
    ``stream()`` takes over — same contract as :meth:`run`, with the
    primary's failure threaded into the fallback's context.  Once
    tokens have been yielded the error propagates instead (the
    partial output cannot be retracted).

    **Timeout.** ``self.timeout`` is the *total* deadline for the
    stream — the same meaning it has in :meth:`run`.  Stall
    detection between chunks is the engine's job
    (``LLMEngine(stream_idle_timeout=...)``).

    **Not applied in streaming** (use :meth:`run` when you need
    them): ``verify=`` and ``output=`` validation — tokens are
    emitted as they arrive, so there is no pre-delivery result to
    judge or validate.

    Multimodal: pass ``images=`` / ``audio=`` to attach blocks to
    the streamed turn — same coercion semantics as :meth:`run`.
    """
    env = self._to_envelope(task, images=images, audio=audio)
    env = self._inject_sources(env)

    # Apply input guard before the first token is emitted.  A blocked
    # task must never reach the provider even in streaming mode.
    if self.guard:
        action = await self.guard.acheck_input(env.task or "")
        if not action.allowed:
            raise ValueError(action.message or "Blocked by guard")
        if action.modified_text is not None:
            env = env.model_copy(update={"task": action.modified_text, "payload": action.modified_text})

    chunks: list[str] = []
    try:
        async for chunk in self._stream_engine(env):
            yield chunk
            chunks.append(chunk)
    except (asyncio.CancelledError, TimeoutError):
        raise
    except Exception as exc:
        # ``getattr`` keeps this safe for Agents built via ``__new__``
        # (test helpers, custom subclasses) that haven't set fallback.
        fallback = getattr(self, "fallback", None)
        if chunks or fallback is None:
            raise
        # Primary engine failed before any token reached the consumer:
        # hand over to the fallback's own stream() pipeline — same
        # contract as run(), with the failure mode threaded into the
        # fallback's context so it can adapt.
        note = f"Previous attempt failed with {type(exc).__name__}: {exc}"
        merged_context = f"{env.context}\n\n{note}" if env.context else note
        fallback_env: Envelope[Any] = Envelope(
            task=env.task,
            context=merged_context,
            images=env.images,
            audio=env.audio,
            payload=env.payload,
        )
        async for chunk in fallback.stream(fallback_env):
            yield chunk
        return  # the fallback ran its own guard / store pipeline

    text = "".join(chunks)
    if self.guard:
        action = await self.guard.acheck_output(text)
        if not action.allowed:
            raise ValueError(action.message or "Output blocked by guard")

    _store = getattr(self, "store", None)
    if _store is not None and getattr(self, "name", None) and chunks:
        from lazybridge.sentinels import _AGENT_OUTPUT_KEY_PREFIX

        _store.write(_AGENT_OUTPUT_KEY_PREFIX + self.name, text)

as_tool

as_tool(name: str | None = None, description: str | None = None, *, verify: Agent | Callable[[str], Any] | None = None, max_verify: int = 3) -> Tool

Wrap this agent as a :class:Tool (advanced / compatibility API).

The canonical way to use a sub-agent is to give it an explicit name= and pass it directly in tools=[...]::

# Canonical
researcher = Agent(name="research", engine=LLMEngine(...))
orchestrator = Agent(..., tools=[researcher])

.as_tool() remains available for local aliases and backward compatibility::

# Advanced alias — use a different name than the agent's own
tools=[researcher.as_tool("deep_research")]

# Backward compat — existing code that already calls as_tool()
tools=[researcher.as_tool("research")]

The tool schema is (task: str) -> Envelope.

Verify (Option B) — wrap the call in a judge/retry loop so every invocation is vetted before returning::

judge = Agent(engine=LLMEngine(
    "claude-opus-4-7",
    system="Reply 'approved' or 'rejected: <reason>'.",
))
synth = Agent(name="synth", engine=LLMEngine(...))
orchestrator = Agent(
    ...,
    tools=[synth.as_tool("synthesize", verify=judge, max_verify=2)],
)

verify can be either an :class:Agent (its run method is called with the output) or a plain callable taking the output text and returning a verdict string / bool. On rejection, the judge's feedback is injected into the next attempt's task.

Source code in lazybridge/agent.py

def as_tool(
    self,
    name: str | None = None,
    description: str | None = None,
    *,
    verify: Agent | Callable[[str], Any] | None = None,
    max_verify: int = 3,
) -> Tool:
    """Wrap this agent as a :class:`Tool` (advanced / compatibility API).

    The canonical way to use a sub-agent is to give it an explicit
    ``name=`` and pass it directly in ``tools=[...]``::

        # Canonical
        researcher = Agent(name="research", engine=LLMEngine(...))
        orchestrator = Agent(..., tools=[researcher])

    ``.as_tool()`` remains available for **local aliases** and
    **backward compatibility**::

        # Advanced alias — use a different name than the agent's own
        tools=[researcher.as_tool("deep_research")]

        # Backward compat — existing code that already calls as_tool()
        tools=[researcher.as_tool("research")]

    The tool schema is ``(task: str) -> Envelope``.

    Verify (Option B) — wrap the call in a judge/retry loop so every
    invocation is vetted before returning::

        judge = Agent(engine=LLMEngine(
            "claude-opus-4-7",
            system="Reply 'approved' or 'rejected: <reason>'.",
        ))
        synth = Agent(name="synth", engine=LLMEngine(...))
        orchestrator = Agent(
            ...,
            tools=[synth.as_tool("synthesize", verify=judge, max_verify=2)],
        )

    ``verify`` can be either an :class:`Agent` (its ``run`` method is
    called with the output) or a plain callable taking the output text
    and returning a verdict string / bool. On rejection, the judge's
    feedback is injected into the next attempt's task.
    """
    if max_verify < 1:
        raise ValueError(f"max_verify must be >= 1, got {max_verify!r}")
    agent = self
    effective_name = name or self.name
    effective_desc = description or self.description or f"Run the {effective_name} agent."

    if verify is None:

        async def _run(task: str) -> Envelope:
            # ``_run_as_tool`` (not ``run``) so a ``conclude`` raised inside
            # the sub-agent propagates up to the top-level caller instead of
            # being absorbed here.  Duck-typed doubles without it fall back
            # to ``run``.
            runner = getattr(agent, "_run_as_tool", agent.run)
            result = await runner(task)
            # Always write under the alias so from_agent("alias") can find
            # the output regardless of agent.name.  _run_body also writes
            # under agent.name (for standalone callers); the alias write
            # here is the authoritative key for Plan sentinel resolution.
            _store = getattr(agent, "store", None)
            if _store is not None and result.ok and effective_name != getattr(agent, "name", None):
                from lazybridge.sentinels import _AGENT_OUTPUT_KEY_PREFIX

                _store.write(_AGENT_OUTPUT_KEY_PREFIX + effective_name, result.text())
            return result
    else:

        async def _run(task: str) -> Envelope:  # type: ignore[misc]
            from lazybridge._verify import verify_with_retry
            from lazybridge.envelope import Envelope as _Env

            env = _Env.from_task(str(task))
            result = await verify_with_retry(
                agent,
                env,
                verify,
                max_verify=max_verify,
            )
            _store = getattr(agent, "store", None)
            if _store is not None and result.ok and effective_name != getattr(agent, "name", None):
                from lazybridge.sentinels import _AGENT_OUTPUT_KEY_PREFIX

                _store.write(_AGENT_OUTPUT_KEY_PREFIX + effective_name, result.text())
            return result

    _run.__name__ = effective_name
    _run.__doc__ = effective_desc

    return Tool(
        _run,
        name=effective_name,
        description=effective_desc,
        mode="signature",
        returns_envelope=True,
        agent_memory=getattr(self, "memory", None),
        agent_store=getattr(self, "store", None),
    )

definition

definition() -> Any

ToolDefinition for this agent — used when passed in tools=[] of another agent.

Source code in lazybridge/agent.py

def definition(self) -> Any:
    """ToolDefinition for this agent — used when passed in tools=[] of another agent."""
    return self.as_tool().definition()

derive

derive(*, tools: list[Any] | None = None, **overrides: Any) -> Agent

Return a NEW Agent: a clone of self plus extra tools and/or overrides.

Does not mutate self. The returned Agent runs full constructor validation, so the build-time guarantee (construction implies a valid graph) is preserved unchanged.

Use when an existing agent needs context-specific capabilities — a pool handle, a one-shot tool, a different name in a sub-context — without rebuilding it field by field.

debater = base.derive(tools=[pool.as_tool(), conclude, vote_tool])
renamed = base.derive(name="alias")

tools= are APPENDED to the agent's existing tools. Any other keyword in overrides replaces the corresponding constructor argument.

Source code in lazybridge/agent.py

def derive(self, *, tools: list[Any] | None = None, **overrides: Any) -> Agent:
    """Return a NEW Agent: a clone of self plus extra tools and/or overrides.

    Does not mutate self. The returned Agent runs full constructor
    validation, so the build-time guarantee (construction implies a valid
    graph) is preserved unchanged.

    Use when an existing agent needs context-specific capabilities — a pool
    handle, a one-shot tool, a different name in a sub-context — without
    rebuilding it field by field.

        debater = base.derive(tools=[pool.as_tool(), conclude, vote_tool])
        renamed = base.derive(name="alias")

    ``tools=`` are APPENDED to the agent's existing tools. Any other keyword
    in ``overrides`` replaces the corresponding constructor argument.
    """
    base_kwargs = dict(
        engine=self.engine,
        tools=list(self._tools_raw) + list(tools or []),
        output=self.output,
        memory=self.memory,
        store=self.store,
        sources=list(self.sources),
        guard=self.guard,
        verify=self.verify,
        max_verify=self.max_verify,
        name=self.name,
        description=self.description,
        session=self.session,
        verbose=self._verbose,
        timeout=self.timeout,
        output_validator=self.output_validator,
        max_output_retries=self.max_output_retries,
        fallback=self.fallback,
    )
    base_kwargs.update(overrides)
    derived = Agent(**base_kwargs)
    # __init__ stamps engine._agent_name on every construction; since the engine
    # object is shared, restore the original so base-agent runs continue to
    # log/emit under the correct identity.
    if hasattr(self.engine, "_agent_name"):
        self.engine._agent_name = self.name
    # Preserve _name_explicit unless name= was explicitly overridden — passing
    # name=self.name through __init__ would otherwise silently promote an
    # implicitly-named agent to explicitly-named, bypassing the guard.
    if "name" not in overrides:
        derived._name_explicit = self._name_explicit
    return derived

from_provider classmethod

from_provider(provider: str, *, tier: Tier | str = 'medium', **kwargs: Any) -> Agent

Construct an Agent for provider using its tier alias for model selection.

Tiers (super_cheap / cheap / medium / expensive / top) resolve to each provider's current lineup, so preview and date-pinned model names stay in one place::

Agent.from_provider("anthropic", tier="top")
Agent.from_provider("openai", tier="cheap", tools=[search])

Source code in lazybridge/agent.py

@classmethod
def from_provider(
    cls,
    provider: str,
    *,
    tier: Tier | str = "medium",
    **kwargs: Any,
) -> Agent:
    """Construct an Agent for ``provider`` using its tier alias for model selection.

    Tiers (``super_cheap`` / ``cheap`` / ``medium`` / ``expensive`` /
    ``top``) resolve to each provider's current lineup, so preview and
    date-pinned model names stay in one place::

        Agent.from_provider("anthropic", tier="top")
        Agent.from_provider("openai", tier="cheap", tools=[search])
    """
    from lazybridge.engines.llm import LLMEngine

    # Pass both the tier (as the provider-facing model string, which
    # the BaseProvider resolves via its tier map) AND the explicit
    # provider name (so _infer_provider doesn't fall back to the
    # default when the tier alone isn't a recognised model).
    return cls(engine=LLMEngine(tier, provider=provider), **kwargs)

chain classmethod

chain(*agents: Agent, **kwargs: Any) -> Agent

Run agents sequentially: output of each becomes input to the next.

Source code in lazybridge/agent.py

@classmethod
def chain(cls, *agents: Agent, **kwargs: Any) -> Agent:
    """Run agents sequentially: output of each becomes input to the next."""
    from lazybridge.engines.plan import Plan, Step

    steps = [Step(target=a, name=a.name) for a in agents]
    plan = Plan(*steps)
    name = kwargs.pop("name", "chain")
    # Don't auto-wrap agents as tools — ``Plan._exec_step`` dispatches
    # Agent targets via ``target.run()`` directly, so wrapping them
    # would just waste schema-compilation on every chain call.
    # Caller-supplied tools= in kwargs still pass through unchanged.
    return cls(engine=plan, name=name, **kwargs)

parallel classmethod

parallel(*agents: Agent, concurrency_limit: int | None = None, step_timeout: float | None = None, **kwargs: Any) -> ParallelAgent

Deterministic fan-out: run agents concurrently on the same task.

Returns a :class:ParallelAgent whose __call__ produces a single :class:Envelope — labelled-text join of every branch's output, with transitive cost rollup. For typed access to per-branch envelopes call ParallelAgent.run_branches(task) (async).

Use this when you know you want N things to happen in parallel. If you want the LLM to decide whether to call agents in parallel (and which, and how), don't use this — pass them as tools=[...] on a regular Agent instead; the engine emits parallel tool calls automatically when the model requests them.

Source code in lazybridge/agent.py

@classmethod
def parallel(
    cls,
    *agents: Agent,
    concurrency_limit: int | None = None,
    step_timeout: float | None = None,
    **kwargs: Any,
) -> ParallelAgent:
    """Deterministic fan-out: run ``agents`` concurrently on the same task.

    Returns a :class:`ParallelAgent` whose ``__call__`` produces a
    single :class:`Envelope` — labelled-text join of every branch's
    output, with transitive cost rollup.  For typed access to per-branch
    envelopes call ``ParallelAgent.run_branches(task)`` (async).

    Use this when you **know** you want N things to happen in
    parallel.  If you want the LLM to decide whether to call agents
    in parallel (and which, and how), don't use this — pass them as
    ``tools=[...]`` on a regular ``Agent`` instead; the engine emits
    parallel tool calls automatically when the model requests them.
    """
    return ParallelAgent(
        agents=list(agents),
        concurrency_limit=concurrency_limit,
        step_timeout=step_timeout,
        **kwargs,
    )

lazybridge.ParallelAgent

ParallelAgent(agents: list[Agent], *, concurrency_limit: int | None = None, step_timeout: float | None = None, name: str = 'parallel', description: str | None = None, session: Any | None = None)

Deterministic fan-out over N agents — the shape behind :meth:Agent.parallel.

Pre-scripted parallel runner. Every input agent receives the same task; the N branch results are folded into a single :class:Envelope via labelled-text join — same shape as :class:Plan's from_parallel_all aggregator. Cost roll-up is transitive. The first non-None branch error propagates as the wrapper's error so downstream consumers can short-circuit.

Prefer :class:Agent with tools=[...] when you want the engine (LLM, Supervisor, Plan) to decide dynamically which tools to invoke and when — parallel execution is automatic on that path.

Per-branch typed access: call :meth:run_branches (async) when you need list[Envelope] rather than the joined wrapper.

Source code in lazybridge/agent.py

def __init__(
    self,
    agents: list[Agent],
    *,
    concurrency_limit: int | None = None,
    step_timeout: float | None = None,
    name: str = "parallel",
    description: str | None = None,
    session: Any | None = None,
) -> None:
    self.agents = agents
    self.concurrency_limit = concurrency_limit
    self.step_timeout = step_timeout
    self.name = name
    self.description = description
    self.session = session

run_branches async

run_branches(task: str | Envelope) -> list[Envelope]

Async per-branch entry point — returns one Envelope per input agent in input order. Use this when you need typed access to individual branch results; for the framework-uniform single-Envelope view, use :meth:run or __call__.

Source code in lazybridge/agent.py

async def run_branches(self, task: str | Envelope) -> list[Envelope]:
    """Async per-branch entry point — returns one ``Envelope`` per
    input agent in input order.  Use this when you need typed
    access to individual branch results; for the framework-uniform
    single-Envelope view, use :meth:`run` or ``__call__``.
    """
    env = Agent._to_envelope(task) if isinstance(task, str) else task
    sem = asyncio.Semaphore(self.concurrency_limit) if self.concurrency_limit else None

    async def _run_one(agent: Agent) -> Envelope:
        async def _coro() -> Envelope:
            if self.step_timeout:
                return await asyncio.wait_for(agent.run(env), timeout=self.step_timeout)
            return await agent.run(env)

        if sem:
            async with sem:
                return await _coro()
        return await _coro()

    results = await asyncio.gather(*[_run_one(a) for a in self.agents], return_exceptions=True)
    out: list[Envelope] = []
    for r in results:
        if isinstance(r, Envelope):
            out.append(r)
        elif isinstance(r, asyncio.CancelledError):
            # CancelledError is BaseException (not Exception) in Python 3.8+;
            # wrapping it as an error envelope would silently swallow the
            # cancellation signal. Re-raise so structured cancellation works.
            raise r
        elif isinstance(r, Exception):
            out.append(Envelope.error_envelope(r))
        else:
            out.append(Envelope.error_envelope(RuntimeError(str(r))))
    return out

run async

run(task: str | Envelope) -> Envelope

Run every branch and return one folded :class:Envelope.

The wrapper's payload is the labelled-text join of every branch's .text(); metadata.nested_* rolls every branch's cost up so the outer envelope reports total spend. The first non-None branch error propagates as the wrapper's error.

For typed per-branch access, call :meth:run_branches.

Source code in lazybridge/agent.py

async def run(self, task: str | Envelope) -> Envelope:
    """Run every branch and return one folded :class:`Envelope`.

    The wrapper's ``payload`` is the labelled-text join of every
    branch's ``.text()``; ``metadata.nested_*`` rolls every branch's
    cost up so the outer envelope reports total spend.  The first
    non-``None`` branch error propagates as the wrapper's ``error``.

    For typed per-branch access, call :meth:`run_branches`.
    """
    branches = await self.run_branches(task)
    return self._join_branches(task, branches)

as_tool

as_tool(name: str | None = None, description: str | None = None) -> Tool

Expose the fan-out runner as a single :class:Tool.

Just delegates to :meth:run — same labelled-text Envelope as every direct caller sees, so a ParallelAgent passed in tools=[...] produces output identical to a hand-call.

Source code in lazybridge/agent.py

def as_tool(
    self,
    name: str | None = None,
    description: str | None = None,
) -> Tool:
    """Expose the fan-out runner as a single :class:`Tool`.

    Just delegates to :meth:`run` — same labelled-text Envelope as
    every direct caller sees, so a ``ParallelAgent`` passed in
    ``tools=[...]`` produces output identical to a hand-call.
    """
    from lazybridge.tools import Tool

    actual_name = name or self.name or "parallel"
    actual_desc = (
        description or self.description or (f"Run {len(self.agents)} agents in parallel and join their outputs.")
    )

    async def _run(task: str) -> Envelope:
        return await self.run(task)

    _run.__name__ = actual_name
    _run.__doc__ = actual_desc

    return Tool(
        _run,
        name=actual_name,
        description=actual_desc,
        mode="signature",
        returns_envelope=True,
    )

lazybridge.Envelope

Bases: BaseModel, Generic[T]

Typed envelope carrying a payload of type T.

Envelope[str] → payload is a string. Envelope[MyModel] → payload is an instance of MyModel. Envelope (no parameter) defaults to T = Any for maximum flexibility.

Multimodal attachments (images= / audio=) ride alongside task and reach the LLMEngine's user-message builder verbatim. Steps in a Plan see them only on step 0; downstream steps receive upstream output (text), not the original attachments.

__str__

__str__() -> str

Stringification falls through to :meth:text.

Needed because tools that return an Envelope (agent-as-tool) cross back into content blocks expected by the LLM API, which serialise the value via str(...). Without this, any such tool would produce "task=… context=…" garbage instead of the agent's actual answer.

Source code in lazybridge/envelope.py

def __str__(self) -> str:
    """Stringification falls through to :meth:`text`.

    Needed because tools that return an ``Envelope`` (agent-as-tool)
    cross back into content blocks expected by the LLM API, which
    serialise the value via ``str(...)``.  Without this, any such
    tool would produce ``"task=…  context=…"`` garbage instead of
    the agent's actual answer.
    """
    return self.text()

Multimodal content blocks

For mixed-modality inputs (text + image + audio), pass images= and audio= kwargs on agent(...), await agent.run(...), or async for chunk in agent.stream(...). Bare URL strings, Path objects, raw bytes, and dict payloads are coerced into the typed blocks below automatically — use these constructors directly only when you need to override the auto-detected MIME type. Narrative coverage lives in Guides → Mid → Multimodal.

lazybridge.ImageContent dataclass

ImageContent(url: str | None = None, base64_data: str | None = None, media_type: str = 'image/jpeg', type: ContentType = ContentType.IMAGE)

from_data_uri classmethod

from_data_uri(data_uri: str) -> ImageContent

Parse data:image/png;base64,<...> style URIs.

Source code in lazybridge/core/types.py

@classmethod
def from_data_uri(cls, data_uri: str) -> ImageContent:
    """Parse ``data:image/png;base64,<...>`` style URIs."""
    media_type, b64 = _parse_data_uri(data_uri, default_mime="image/jpeg")
    return cls(base64_data=b64, media_type=media_type)

lazybridge.AudioContent dataclass

AudioContent(url: str | None = None, base64_data: str | None = None, media_type: str = 'audio/wav', type: ContentType = ContentType.AUDIO)

Audio attachment for multimodal LLM input.

Provider support varies by model — see :meth:BaseProvider.supports_audio. Anthropic / OpenAI accept base64 only; Google Gemini accepts both URL and base64.

from_data_uri classmethod

from_data_uri(data_uri: str) -> AudioContent

Parse data:audio/flac;base64,<...> style URIs.

Source code in lazybridge/core/types.py

@classmethod
def from_data_uri(cls, data_uri: str) -> AudioContent:
    """Parse ``data:audio/flac;base64,<...>`` style URIs."""
    media_type, b64 = _parse_data_uri(data_uri, default_mime="audio/wav")
    return cls(base64_data=b64, media_type=media_type)