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

    # 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

    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.
                agent_raw.session = self.session
                _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.
        # ``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 immediately; a modified input (GuardAction.modify) replaces the task in the envelope sent to the provider. This is identical to the guard contract in :meth:run — streaming mode does not bypass the guard.

Honours self.timeout between chunks so a stalled provider can't hang the caller. Each __anext__ is wrapped in asyncio.wait_for (per-chunk, not whole-stream) so short chunks don't consume the deadline budget.

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` immediately; a modified input
    (``GuardAction.modify``) replaces the task in the envelope sent
    to the provider.  This is identical to the guard contract in
    :meth:`run` — streaming mode does not bypass the guard.

    Honours ``self.timeout`` between chunks so a stalled provider
    can't hang the caller.  Each ``__anext__`` is wrapped in
    ``asyncio.wait_for`` (per-chunk, not whole-stream) so short
    chunks don't consume the deadline budget.

    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)
    timeout = getattr(self, "timeout", None)

    # 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})

    gen = self.engine.stream(
        env,
        tools=list(self._tool_map.values()),
        output_type=self.output,
        memory=self.memory,
        session=self.session,
    ).__aiter__()
    chunks: list[str] = []
    _completed = False
    try:
        while True:
            try:
                if timeout is None:
                    chunk = await gen.__anext__()
                else:
                    chunk = await asyncio.wait_for(gen.__anext__(), timeout=timeout)
            except StopAsyncIteration:
                _completed = True
                return
            yield chunk
            chunks.append(chunk)
    finally:
        aclose = getattr(gen, "aclose", None)
        if aclose is not None:
            try:
                await aclose()
            except asyncio.CancelledError:
                # Cancellation is BaseException; let it propagate so
                # structured cancellation works through the boundary.
                raise
            except Exception as exc:
                # Surface non-cancellation errors as a UserWarning so
                # buggy provider cleanup paths don't disappear; we
                # still don't re-raise (the consumer has either
                # finished or already abandoned the stream).
                import warnings as _w

                _w.warn(
                    f"stream() aclose raised {type(exc).__name__}: {exc}.",
                    stacklevel=2,
                )
        if _completed:
            _store = getattr(self, "store", None)
            if _store is not None and self.name and chunks:
                from lazybridge.sentinels import _AGENT_OUTPUT_KEY_PREFIX

                _store.write(_AGENT_OUTPUT_KEY_PREFIX + self.name, "".join(chunks))

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."""
    if not data_uri.startswith("data:"):
        raise ValueError("expected a data: URI")
    header, _, body = data_uri[5:].partition(",")
    if not body:
        raise ValueError("malformed data URI: missing payload")
    media_type, _, encoding = header.partition(";")
    media_type = media_type or "image/jpeg"
    if encoding == "base64":
        return cls(base64_data=body, media_type=media_type)
    return cls(base64_data=base64.b64encode(body.encode()).decode("ascii"), 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."""
    if not data_uri.startswith("data:"):
        raise ValueError("expected a data: URI")
    header, _, body = data_uri[5:].partition(",")
    if not body:
        raise ValueError("malformed data URI: missing payload")
    media_type, _, encoding = header.partition(";")
    media_type = media_type or "audio/wav"
    if encoding == "base64":
        return cls(base64_data=body, media_type=media_type)
    return cls(base64_data=base64.b64encode(body.encode()).decode("ascii"), media_type=media_type)