API reference¶

Autogenerated from docstrings.

Safety¶

lazytools.safety.Allowlist ¶

Allowlist(allowed: Iterable[object] | None)

Case-insensitive, string-normalized target allow-list.

None means "no allow-list configured" → permits everything. An empty iterable means "deny everything".

Source code in src/lazytools/safety/allowlist.py

def __init__(self, allowed: Iterable[object] | None) -> None:
    self._allowed = None if allowed is None else {str(a).lower() for a in allowed}

lazytools.safety.ConfirmationGate ¶

ConfirmationGate(*, enabled: bool = True)

One-shot, target-bound confirmation grants for dangerous actions.

Not a sticky boolean: each grant authorizes exactly one action, so an approved single message can never silently authorize a flood. Grants are matched from most to least specific: a target+scope grant before a target-only one, then an any-target+scope grant before an any-target one. A scope-bound grant is never spendable when no scope (None) is supplied at consume time. No process-global mutable state — grants live on the instance.

The gate is scope-agnostic: the caller decides what scope means (in LazyPulse it is the running task id, read from :func:lazytools.safety.current_scope) and passes it in. This is what keeps the safety layer free of any orchestration dependency.

Source code in src/lazytools/safety/gates.py

def __init__(self, *, enabled: bool = True) -> None:
    self._enabled = enabled
    # Keys are ``(target, scope)`` where target is a lowercased string or
    # ``_ANY`` and scope is an opaque binding (the task id) or ``None``.
    self._grants: dict[tuple[str, str | None], int] = {}
    # ``grant``/``confirm_*`` typically run on a review-queue/UI thread
    # while a worker consumes; the lock makes grant/consume atomic so a
    # single grant can never be double-spent across threads.
    self._mutex = threading.Lock()

grant ¶

grant(target: object, *, scope: str | None = None) -> None

Authorize exactly one action to target (the tighter grant).

Source code in src/lazytools/safety/gates.py

def grant(self, target: object, *, scope: str | None = None) -> None:
    """Authorize exactly one action to ``target`` (the tighter grant)."""
    self._add((str(target).lower(), scope))

grant_any ¶

grant_any(*, scope: str | None = None) -> None

Authorize exactly one action to any target (subject to allow-list).

Source code in src/lazytools/safety/gates.py

def grant_any(self, *, scope: str | None = None) -> None:
    """Authorize exactly one action to any target (subject to allow-list)."""
    self._add((_ANY, scope))

consume ¶

consume(target: object, *, scope: str | None = None) -> bool

Spend one matching grant for target in scope; True if found.

Returns True immediately when the gate is disabled. A scope-bound grant is only matched when the same scope is supplied here.

Source code in src/lazytools/safety/gates.py

def consume(self, target: object, *, scope: str | None = None) -> bool:
    """Spend one matching grant for ``target`` in ``scope``; ``True`` if found.

    Returns ``True`` immediately when the gate is disabled. A scope-bound
    grant is only matched when the same ``scope`` is supplied here.
    """
    if not self._enabled:
        return True
    target_l = str(target).lower()
    candidates: list[tuple[str, str | None]] = []
    if scope is not None:
        candidates.append((target_l, scope))
    candidates.append((target_l, None))
    if scope is not None:
        candidates.append((_ANY, scope))
    candidates.append((_ANY, None))
    with self._mutex:
        for key in candidates:
            if self._grants.get(key, 0) > 0:
                self._grants[key] -= 1
                return True
    return False

lazytools.safety.ActionBlocked ¶

Bases: PermissionError

Base for dangerous-action denials (allow-list / confirmation).

Subclasses PermissionError so existing except PermissionError handlers keep working. Carries an audit-friendly message that names the action and the reason and never leaks secrets.

lazytools.safety.current_scope ¶

current_scope() -> str | None

Return the ambient scope of the current run, if any.

Source code in src/lazytools/safety/context.py

def current_scope() -> str | None:
    """Return the ambient scope of the current run, if any."""
    return active_scope.get()

lazytools.safety.UrlBlocked ¶

Bases: ActionBlocked

Raised when a URL fails the SSRF guard (scheme / host / IP check).

lazytools.safety.validate_public_url ¶

validate_public_url(url: str, *, allowed_hosts: Collection[str] | None = None) -> str

Validate that url is a public http(s) URL; return it unchanged.

Parameters:

Name	Type	Description	Default
`url`	`str`	The absolute URL about to be fetched (including redirect targets).	required
`allowed_hosts`	`Collection[str] \| None`	Optional set of permitted hostnames (compared case-insensitively). `None` skips the host pinning check.	`None`

Raises:

Type	Description
`UrlBlocked`	On a non-http(s) scheme, a missing hostname, a hostname outside `allowed_hosts`, or a non-global literal IP host (loopback, private, link-local, multicast, reserved, unspecified).

Source code in src/lazytools/safety/urls.py

def validate_public_url(url: str, *, allowed_hosts: Collection[str] | None = None) -> str:
    """Validate that ``url`` is a public http(s) URL; return it unchanged.

    Args:
        url: The absolute URL about to be fetched (including redirect targets).
        allowed_hosts: Optional set of permitted hostnames (compared
            case-insensitively). ``None`` skips the host pinning check.

    Raises:
        UrlBlocked: On a non-http(s) scheme, a missing hostname, a hostname
            outside ``allowed_hosts``, or a non-global literal IP host
            (loopback, private, link-local, multicast, reserved, unspecified).
    """
    parts = urlsplit(url)
    if parts.scheme not in ("http", "https"):
        raise UrlBlocked(f"refused URL {url!r}: scheme {parts.scheme!r} is not http(s)")
    host = parts.hostname
    if not host:
        raise UrlBlocked(f"refused URL {url!r}: missing hostname")
    host = host.lower()
    if allowed_hosts is not None and host not in {h.lower() for h in allowed_hosts}:
        raise UrlBlocked(f"refused URL {url!r}: host {host!r} is not an allowed host")
    try:
        ip: ipaddress.IPv4Address | ipaddress.IPv6Address = ipaddress.ip_address(host)
    except ValueError:
        legacy = _legacy_ipv4_literal(host)
        if legacy is None:
            # A DNS name, not a literal IP. Pinning via ``allowed_hosts`` is the
            # real control for names; nothing more to check syntactically.
            return url
        ip = legacy
    if not ip.is_global:
        raise UrlBlocked(f"refused URL {url!r}: IP {host!r} is not globally routable")
    return url

Artifact registry¶

lazytools.registry.DBEntry `dataclass` ¶

DBEntry(name: str, env_var: str, owner_repo: str, required: bool = True, description: str = '')

One known ecosystem DB: which env var holds its path, who owns it.

Attributes:

Name	Type	Description
`name`	`str`	Stable logical identifier (e.g. `"market_data"`).
`env_var`	`str`	Name of the environment variable holding the DB path/DSN.
`owner_repo`	`str`	The repo that owns and writes this DB.
`required`	`bool`	If `True`, a caller expects this DB to always be configured in a fully set-up deployment; :func:`resolve_db` raises `RuntimeError` rather than returning `None` when it is unset. Artifact DBs (opt-in, per-repo) are `required=False`.
`description`	`str`	Short human-readable note on what the DB holds.

lazytools.registry.KNOWN_DBS `module-attribute` ¶

KNOWN_DBS: tuple[DBEntry, ...] = (DBEntry('market_data', 'MARKET_DATA_DB', 'market-data-hub', True, 'Prices and historical series'), DBEntry('pulse_state', 'STORE_DB', 'lazypulse', False, "LazyPulse's always-on PulseAgent/Telegram bot state store. Optional, not required: a deployment that schedules its jobs externally (e.g. the Windows Task Scheduler) runs no PulseAgent and so has nothing to persist here. It was declared required=True, which made status() report a missing required DB in every such deployment -- a permanent false alarm that trains the reader to ignore the one signal the registry exists to give. No caller resolves this entry; set STORE_DB only when actually running the always-on agent."), DBEntry('crawler_raw', 'LAZYCRAWLER_NEWS_DB', 'lazycrawler', True, 'News crawl page cache'), DBEntry('lazystats_depot', 'LAZYSTATS_RESULT_DEPOT_DB', 'lazystats', True, 'LazyStats analysis result depot (regime, regression, ...)'), DBEntry('regime_tools_db', 'LAZYTOOLS_REGIME_DB', 'lazystats', False, "LazyHMM regime-fitting tool depot (fitted params, figures, state sequences) backing LazyTools' regime_* MCP tools -- a separate store from lazystats_depot, which holds market-data-hub's persisted regime *run results*, not the fitting tools' own state"), DBEntry('market_data_artifacts', 'MARKET_DATA_ARTIFACTS_DB', 'market-data-hub', False, 'Artifacts produced by market-data-hub'), DBEntry('pulse_artifacts', 'PULSE_ARTIFACTS_DB', 'lazypulse', False, 'Artifacts produced by LazyPulse'), DBEntry('crawler_artifacts', 'CRAWLER_ARTIFACTS_DB', 'lazycrawler', False, 'Artifacts produced by LazyCrawler'), DBEntry('crawler_econ_state', 'ECON_STATE_DB', 'lazycrawler', False, 'Economic-release monitor cursor state (which releases have already been reported). NOT append-only history: two copies each advance their own cursor, so never union them -- take the one the live producer most recently advanced.'), DBEntry('crawler_digests', 'DIGESTS_DB', 'lazycrawler', False, 'Full text of every executive news digest (make_news_report.py), keyed UNIQUE(session_id, engine) -- crawler_artifacts holds the catalogue entry and file pointer, this holds the prose itself. make_news_report.py reads this variable and only falls back to a checkout-relative reports/news/digests.db when it is unset, which is how a pinned runtime worktree once split the history.'), DBEntry('lazyray_db', 'LAZYRAY_DB', 'lazyray', False, "LazyRay's own DuckDB output (Dalio-style scores, regimes, classifications). LazyRay resolves its own settings-based default when this is unset -- that silent fallback split a deployment's history once, so deployments should wire it explicitly."), DBEntry('lazyportfolio_artifacts', 'LAZYPORTFOLIO_ARTIFACTS_DB', 'lazyportfolio', False, 'Artifacts (reports) produced by LazyPortfolio'), DBEntry('lazyportfolio_store', 'LAZYPORTFOLIO_TREE_DB', 'lazyportfolio', False, "Tree Studio's primary store (lazyportfolio.v2.db): saved tree configs, and structured run history/artifacts (weights, metrics, data-as-of, config hash) for every estimate/backtest/report run. A separate store from lazyportfolio_artifacts above: that one is the opt-in cross-repo artifact catalog entry for the rendered HTML report; this one is LazyPortfolio's own primary data. Like every optional entry here, resolve_db() only returns a path when LAZYPORTFOLIO_TREE_DB is actually set -- LazyPortfolio itself still works with it unset (falls back to a repo-relative default path), but that fallback path is NOT visible through resolve_db()/status(); call lazyportfolio.v2.store.resolve_store_path() directly to see the path actually in use in that case."), DBEntry('anomaly_explanations', 'ANOMALY_EXPLANATIONS_DB', 'lazystats', False, "LLM-generated causal explanations for statistical anomalies (return outliers, volatility/correlation shifts) flagged in lazystats_depot's etf_daily_stats series, plus the Saturday weekly review that verifies them and looks for emerging trends -- narrative/evidence content, kept in its own store separate from lazystats_depot's deterministic quantitative results."))

lazytools.registry.resolve_db ¶

resolve_db(name: str) -> str | None

Resolve a known DB's path from its declared environment variable.

Parameters:

Name	Type	Description	Default
`name`	`str`	Logical DB name — must be one of `KNOWN_DBS`' `name` values.	required

Returns:

Type	Description
`str \| None`	The env var's value (the DB path/DSN) if set. `None` if the entry
`str \| None`	is optional (`required=False`) and its env var is unset.

Raises:

Type	Description
`KeyError`	`name` is not a known DB (not in `KNOWN_DBS`).
`RuntimeError`	The entry is required (`required=True`) and its env var is unset. The message names the env var so the operator knows exactly what to set.

Source code in src/lazytools/registry/db.py

def resolve_db(name: str) -> str | None:
    """Resolve a known DB's path from its declared environment variable.

    Args:
        name: Logical DB name — must be one of ``KNOWN_DBS``' ``name`` values.

    Returns:
        The env var's value (the DB path/DSN) if set. ``None`` if the entry
        is optional (``required=False``) and its env var is unset.

    Raises:
        KeyError: ``name`` is not a known DB (not in ``KNOWN_DBS``).
        RuntimeError: The entry is required (``required=True``) and its env
            var is unset. The message names the env var so the operator
            knows exactly what to set.
    """
    try:
        entry = _BY_NAME[name]
    except KeyError:
        raise KeyError(f"Unknown DB {name!r}. Known DBs: {sorted(_BY_NAME)}") from None

    value = os.environ.get(entry.env_var)
    if value:
        return value
    if entry.required:
        raise RuntimeError(
            f"DB {entry.name!r} (owned by {entry.owner_repo}) requires env var "
            f"{entry.env_var!r} to be set, but it is unset."
        )
    return None

lazytools.registry.status ¶

status() -> list[dict]

Report, for every known DB, whether its env var is currently set.

Returns:

Type	Description
`list[dict]`	One dict per :data:`KNOWN_DBS` entry:
`list[dict]`	`{name, env_var, owner_repo, required, set}`.

Source code in src/lazytools/registry/db.py

def status() -> list[dict]:
    """Report, for every known DB, whether its env var is currently set.

    Returns:
        One dict per :data:`KNOWN_DBS` entry:
        ``{name, env_var, owner_repo, required, set}``.
    """
    return [
        {
            "name": entry.name,
            "env_var": entry.env_var,
            "owner_repo": entry.owner_repo,
            "required": entry.required,
            "set": bool(os.environ.get(entry.env_var)),
        }
        for entry in KNOWN_DBS
    ]

lazytools.registry.artifact_dbs ¶

artifact_dbs() -> list[tuple[str, str]]

List the artifact DBs that are actually configured in this environment.

Returns:

Type	Description
`list[tuple[str, str]]`	`[(owner_repo, path), ...]` for every `KNOWN_DBS` entry whose
`list[tuple[str, str]]`	`name` ends in `"_artifacts"` and whose env var is set. Entries
`list[tuple[str, str]]`	whose env var is unset are silently skipped (artifact DBs are
`list[tuple[str, str]]`	opt-in per repo).

Source code in src/lazytools/registry/db.py

def artifact_dbs() -> list[tuple[str, str]]:
    """List the artifact DBs that are actually configured in this environment.

    Returns:
        ``[(owner_repo, path), ...]`` for every ``KNOWN_DBS`` entry whose
        ``name`` ends in ``"_artifacts"`` and whose env var is set. Entries
        whose env var is unset are silently skipped (artifact DBs are
        opt-in per repo).
    """
    result: list[tuple[str, str]] = []
    for entry in KNOWN_DBS:
        if not entry.name.endswith("_artifacts"):
            continue
        value = os.environ.get(entry.env_var)
        if value:
            result.append((entry.owner_repo, value))
    return result

lazytools.registry.register_artifact ¶

register_artifact(db_path: str, *, repo: str, kind: str, title: str, summary: str, tags: list[str] | None = None, content: str | None = None, content_uri: str | None = None, ttl_days: int | None = None) -> str

Create (if missing) the artifacts table and insert one record.

Parameters:

Name	Type	Description	Default
`db_path`	`str`	Path to the SQLite file (the caller's artifact DB).	required
`repo`	`str`	Which repo produced this artifact (e.g. `"lazytools"`).	required
`kind`	`str`	Free-text artifact category (e.g. `"backtest_report"`).	required
`title`	`str`	Short human-readable title.	required
`summary`	`str`	Cheap-to-read summary — what :func:`search_artifacts` returns; keep the full payload out of this field.	required
`tags`	`list[str] \| None`	Optional list of string tags for filtering.	`None`
`content`	`str \| None`	Optional full payload, stored inline.	`None`
`content_uri`	`str \| None`	Optional pointer to the payload when it lives elsewhere (e.g. an object-storage URL) instead of inline.	`None`
`ttl_days`	`int \| None`	Optional time-to-live in days from now; sets `expires_at`. `None` means the artifact never expires.	`None`

Returns:

Type	Description
`str`	The new artifact's id (`str(uuid.uuid4())`).

Source code in src/lazytools/registry/artifacts.py

def register_artifact(
    db_path: str,
    *,
    repo: str,
    kind: str,
    title: str,
    summary: str,
    tags: list[str] | None = None,
    content: str | None = None,
    content_uri: str | None = None,
    ttl_days: int | None = None,
) -> str:
    """Create (if missing) the ``artifacts`` table and insert one record.

    Args:
        db_path: Path to the SQLite file (the caller's artifact DB).
        repo: Which repo produced this artifact (e.g. ``"lazytools"``).
        kind: Free-text artifact category (e.g. ``"backtest_report"``).
        title: Short human-readable title.
        summary: Cheap-to-read summary — what :func:`search_artifacts`
            returns; keep the full payload out of this field.
        tags: Optional list of string tags for filtering.
        content: Optional full payload, stored inline.
        content_uri: Optional pointer to the payload when it lives
            elsewhere (e.g. an object-storage URL) instead of inline.
        ttl_days: Optional time-to-live in days from now; sets
            ``expires_at``. ``None`` means the artifact never expires.

    Returns:
        The new artifact's id (``str(uuid.uuid4())``).
    """
    artifact_id = str(uuid.uuid4())
    created_at = _now_iso()
    expires_at = None
    if ttl_days is not None:
        expires_at = (datetime.now(UTC) + timedelta(days=ttl_days)).isoformat()
    content_hash = hashlib.sha256(content.encode("utf-8")).hexdigest() if content is not None else None

    with _connect_write(db_path) as conn:
        conn.execute(
            """
            INSERT INTO artifacts (
                artifact_id, repo, kind, title, summary, tags,
                created_at, expires_at, content, content_uri, content_hash
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                artifact_id,
                repo,
                kind,
                title,
                summary,
                json.dumps(tags or []),
                created_at,
                expires_at,
                content,
                content_uri,
                content_hash,
            ),
        )
    return artifact_id

lazytools.registry.search_artifacts ¶

search_artifacts(db_path: str, *, query: str | None = None, kind: str | None = None, tags: list[str] | None = None, since: str | None = None, limit: int = 20) -> list[dict]

Search the artifact catalog. Never returns content — metadata + summary only; call :func:get_artifact for the full record.

Parameters:

Name	Type	Description	Default
`db_path`	`str`	Path to the SQLite file.	required
`query`	`str \| None`	Case-insensitive substring match against `title`, `summary`, or `tags`.	`None`
`kind`	`str \| None`	Exact match on `kind`.	`None`
`tags`	`list[str] \| None`	Every tag in this list must be present on the artifact's tags.	`None`
`since`	`str \| None`	ISO8601 timestamp; only artifacts with `created_at >= since`.	`None`
`limit`	`int`	Maximum rows to return. Must be a positive integer.	`20`

Returns:

Type	Description
`list[dict]`	Matching records (no `content`/`content_uri`/`content_hash`),
`list[dict]`	ordered by `created_at` descending, excluding expired artifacts.

Raises:

Type	Description
`ValueError`	`limit` is not a positive integer.

Source code in src/lazytools/registry/artifacts.py

def search_artifacts(
    db_path: str,
    *,
    query: str | None = None,
    kind: str | None = None,
    tags: list[str] | None = None,
    since: str | None = None,
    limit: int = 20,
) -> list[dict]:
    """Search the artifact catalog. Never returns ``content`` — metadata +
    summary only; call :func:`get_artifact` for the full record.

    Args:
        db_path: Path to the SQLite file.
        query: Case-insensitive substring match against ``title``,
            ``summary``, or ``tags``.
        kind: Exact match on ``kind``.
        tags: Every tag in this list must be present on the artifact's tags.
        since: ISO8601 timestamp; only artifacts with ``created_at >= since``.
        limit: Maximum rows to return. Must be a positive integer.

    Returns:
        Matching records (no ``content``/``content_uri``/``content_hash``),
        ordered by ``created_at`` descending, excluding expired artifacts.

    Raises:
        ValueError: ``limit`` is not a positive integer.
    """
    limit = int(limit)
    if limit < 1:
        raise ValueError(f"limit must be a positive integer, got {limit!r}")

    conn = _connect_read(db_path)
    if conn is None:
        return []

    with conn:
        clauses = ["(expires_at IS NULL OR expires_at > ?)"]
        params: list[object] = [_now_iso()]

        if kind:
            clauses.append("kind = ?")
            params.append(kind)
        if since:
            # created_at is always stored as a UTC isoformat string; a valid
            # non-UTC offset in `since` (e.g. "...+02:00") must compare as
            # the same instant, not as mismatched string spellings.
            since_dt = datetime.fromisoformat(since)
            if since_dt.tzinfo is None:
                since_dt = since_dt.replace(tzinfo=UTC)
            clauses.append("created_at >= ?")
            params.append(since_dt.astimezone(UTC).isoformat())

        where = " AND ".join(clauses)
        query_lower = query.lower() if query else None
        wanted = set(tags) if tags else None

        def _matches(record: dict) -> bool:
            # Matched against the *decoded* tag values, not the raw
            # json.dumps() bytes -- a tag like "café" is escaped to
            # "café" (and one containing quotes gains backslashes) in
            # the serialized column, so a literal-text query would never
            # find it there even though it's an exact tag match.
            if query_lower is not None and not (
                query_lower in record["title"].lower()
                or query_lower in record["summary"].lower()
                or any(query_lower in t.lower() for t in record["tags"])
            ):
                return False
            return wanted is None or wanted.issubset(set(record["tags"]))

        if query_lower is None and wanted is None:
            # No Python-side filter needed -- bound the fetch in SQL itself
            # instead of pulling the whole catalog and discarding excess.
            sql = (
                f"SELECT {', '.join(_METADATA_COLUMNS)} FROM artifacts "
                f"WHERE {where} ORDER BY created_at DESC LIMIT ?"
            )
            rows = conn.execute(sql, [*params, limit]).fetchall()
            return [_row_to_dict(row, _METADATA_COLUMNS) for row in rows]

        # `query`/`tags` need a Python-side check (substring match against
        # decoded tags, or a tag-set subset check), so SQL can't bound the
        # result directly. Fetch in growing batches until `limit` matches
        # are found or the table is exhausted, rather than unconditionally
        # decoding every row.
        matched: list[dict] = []
        batch_size = max(limit * 5, 100)
        offset = 0
        while len(matched) < limit:
            sql = (
                f"SELECT {', '.join(_METADATA_COLUMNS)} FROM artifacts "
                f"WHERE {where} ORDER BY created_at DESC LIMIT ? OFFSET ?"
            )
            rows = conn.execute(sql, [*params, batch_size, offset]).fetchall()
            if not rows:
                break
            for row in rows:
                record = _row_to_dict(row, _METADATA_COLUMNS)
                if _matches(record):
                    matched.append(record)
                    if len(matched) >= limit:
                        break
            if len(rows) < batch_size:
                break  # exhausted the table
            offset += batch_size
        return matched[:limit]

lazytools.registry.get_artifact ¶

get_artifact(db_path: str, artifact_id: str) -> dict | None

Fetch one artifact's full record, including content.

Parameters:

Name	Type	Description	Default
`db_path`	`str`	Path to the SQLite file.	required
`artifact_id`	`str`	The artifact's id.	required

Returns:

Type	Description
`dict \| None`	The full record, or `None` if not found or if its `expires_at`
`dict \| None`	has already passed.

Source code in src/lazytools/registry/artifacts.py

def get_artifact(db_path: str, artifact_id: str) -> dict | None:
    """Fetch one artifact's full record, including ``content``.

    Args:
        db_path: Path to the SQLite file.
        artifact_id: The artifact's id.

    Returns:
        The full record, or ``None`` if not found or if its ``expires_at``
        has already passed.
    """
    conn = _connect_read(db_path)
    if conn is None:
        return None

    with conn:
        row = conn.execute(
            f"SELECT {', '.join(_ROW_COLUMNS)} FROM artifacts WHERE artifact_id = ?",
            (artifact_id,),
        ).fetchone()

    if row is None:
        return None

    record = _row_to_dict(row, _ROW_COLUMNS)
    if record["expires_at"] and record["expires_at"] <= _now_iso():
        return None
    return record

lazytools.registry.search_everywhere ¶

search_everywhere(*, query: str | None = None, kind: str | None = None, tags: list[str] | None = None, since: str | None = None, limit: int = 20) -> list[dict]

Search every configured repo's artifact DB and merge the results.

Parameters:

Name	Type	Description	Default
`query`	`str \| None`	Case-insensitive substring match against title/summary/tags.	`None`
`kind`	`str \| None`	Exact match on artifact kind.	`None`
`tags`	`list[str] \| None`	Every tag in this list must be present on the artifact's tags.	`None`
`since`	`str \| None`	ISO8601 timestamp lower bound on `created_at`.	`None`
`limit`	`int`	Maximum merged rows to return.	`20`

Returns:

Type	Description
`list[dict]`	Records from every artifact DB whose env var is currently set (see
`list[dict]`	func:`lazytools.registry.db.artifact_dbs`), merged, each carrying a
`list[dict]`	`"repo"` field, sorted by `created_at` descending and truncated
`list[dict]`	to `limit`.

Source code in src/lazytools/registry/router.py

def search_everywhere(
    *,
    query: str | None = None,
    kind: str | None = None,
    tags: list[str] | None = None,
    since: str | None = None,
    limit: int = 20,
) -> list[dict]:
    """Search every configured repo's artifact DB and merge the results.

    Args:
        query: Case-insensitive substring match against title/summary/tags.
        kind: Exact match on artifact kind.
        tags: Every tag in this list must be present on the artifact's tags.
        since: ISO8601 timestamp lower bound on ``created_at``.
        limit: Maximum merged rows to return.

    Returns:
        Records from every artifact DB whose env var is currently set (see
        :func:`lazytools.registry.db.artifact_dbs`), merged, each carrying a
        ``"repo"`` field, sorted by ``created_at`` descending and truncated
        to ``limit``.
    """
    merged: list[dict] = []
    for repo, path in db.artifact_dbs():
        records = search_artifacts(path, query=query, kind=kind, tags=tags, since=since, limit=limit)
        for record in records:
            record["repo"] = repo
        merged.extend(records)

    merged.sort(key=lambda r: r["created_at"], reverse=True)
    return merged[:limit]

lazytools.registry.get_everywhere ¶

get_everywhere(repo: str, artifact_id: str) -> dict | None

Fetch one artifact's full record from a specific repo's artifact DB.

Parameters:

Name	Type	Description	Default
`repo`	`str`	The owning repo, as it appears in :func:`lazytools.registry.db.artifact_dbs`' `owner_repo` (e.g. `"market-data-hub"`).	required
`artifact_id`	`str`	The artifact's id.	required

Returns:

Type	Description
`dict \| None`	The full record (see :func:`lazytools.registry.artifacts.get_artifact`),
`dict \| None`	or `None` if the repo has no configured artifact DB, or the
`dict \| None`	artifact is not found/expired.

Source code in src/lazytools/registry/router.py

def get_everywhere(repo: str, artifact_id: str) -> dict | None:
    """Fetch one artifact's full record from a specific repo's artifact DB.

    Args:
        repo: The owning repo, as it appears in
            :func:`lazytools.registry.db.artifact_dbs`' ``owner_repo`` (e.g.
            ``"market-data-hub"``).
        artifact_id: The artifact's id.

    Returns:
        The full record (see :func:`lazytools.registry.artifacts.get_artifact`),
        or ``None`` if the repo has no configured artifact DB, or the
        artifact is not found/expired.
    """
    for candidate_repo, path in db.artifact_dbs():
        if candidate_repo == repo:
            return get_artifact(path, artifact_id)
    return None

lazytools.registry.RegistryTools ¶

RegistryTools(*, allow_write: bool = False)

A ToolProvider exposing the DB registry and artifact catalog.

Parameters:

Name	Type	Description	Default
`allow_write`	`bool`	If `False` (the default), only the read-only tools (`registry_status`, `artifact_search`, `artifact_get`) are emitted. `artifact_register` is a write and is only emitted when this is `True`.	`False`

Source code in src/lazytools/registry/tools.py

def __init__(self, *, allow_write: bool = False) -> None:
    self._allow_write = allow_write

Gmail¶

lazytools.connectors.gmail.GmailTools ¶

GmailTools(client: GmailService, *, allowed_recipients: list[str] | None = None, require_confirmation: bool = True)

A ToolProvider wrapping a :class:GmailService for the worker.

Exposes four tools: gmail_list_emails, gmail_get_email, gmail_create_draft, and gmail_send.

The underlying :class:~lazytools.connectors.gmail.client.GmailClient is thread-safe (serialises calls through an internal lock), so all four tools are safe to invoke from concurrent PulseAgent task workers.

Source code in src/lazytools/connectors/gmail/tools.py

def __init__(
    self,
    client: GmailService,
    *,
    allowed_recipients: list[str] | None = None,
    require_confirmation: bool = True,
) -> None:
    self._client = client
    self._allowlist = Allowlist(allowed_recipients)
    self._gate = ConfirmationGate(enabled=require_confirmation)

require_confirmation `property` ¶

require_confirmation: bool

Whether a send needs an outstanding confirmation (public attribute).

confirm_once ¶

confirm_once(*, task_id: str | None = None) -> None

Authorize exactly one send to any recipient (subject to the allow-list). Call once per approved message. Pass task_id= to bind the grant to a single task so a concurrent task cannot consume it.

Source code in src/lazytools/connectors/gmail/tools.py

def confirm_once(self, *, task_id: str | None = None) -> None:
    """Authorize exactly one send to any recipient (subject to the
    allow-list). Call once per approved message. Pass ``task_id=`` to bind
    the grant to a single task so a concurrent task cannot consume it."""
    self._gate.grant_any(scope=task_id)

confirm_send ¶

confirm_send(*, to: str, task_id: str | None = None) -> None

Authorize exactly one send to a specific recipient — the tighter, preferred grant. Pass task_id= to also bind it to a single task.

Source code in src/lazytools/connectors/gmail/tools.py

def confirm_send(self, *, to: str, task_id: str | None = None) -> None:
    """Authorize exactly one send to a specific recipient — the tighter,
    preferred grant. Pass ``task_id=`` to also bind it to a single task."""
    self._gate.grant(to, scope=task_id)

lazytools.connectors.gmail.GmailClient ¶

GmailClient(service: Any)

Production :class:GmailService backed by googleapiclient.

All methods acquire a per-instance lock before touching the underlying googleapiclient resource so the client is safe to call from multiple threads concurrently (e.g. parallel PulseAgent task workers).

Source code in src/lazytools/connectors/gmail/client.py

def __init__(self, service: Any) -> None:
    # ``service`` is a googleapiclient Resource (or any object exposing
    # the same ``users().messages()`` shape).
    self._service = service
    self._lock = threading.Lock()

from_credentials `classmethod` ¶

from_credentials(*, credentials_path: str, token_path: str, scopes: list[str]) -> GmailClient

Build a client from an OAuth client-secret + cached token file.

Imports the Google libraries lazily; raises a friendly ImportError if the gmail extra is not installed.

Source code in src/lazytools/connectors/gmail/client.py

@classmethod
def from_credentials(
    cls,
    *,
    credentials_path: str,
    token_path: str,
    scopes: list[str],
) -> GmailClient:
    """Build a client from an OAuth client-secret + cached token file.

    Imports the Google libraries lazily; raises a friendly
    ``ImportError`` if the ``gmail`` extra is not installed.
    """
    try:
        from google.auth.transport.requests import Request
        from google.oauth2.credentials import Credentials
        from google_auth_oauthlib.flow import InstalledAppFlow
        from googleapiclient.discovery import build
    except ImportError as exc:  # pragma: no cover — exercised only without the extra
        raise ImportError(
            "GmailClient.from_credentials requires the 'gmail' extra. "
            'Install it with: pip install "lazytoolkit[gmail] @ git+https://github.com/selvaz/LazyTools.git"'
        ) from exc

    import os

    creds = None
    if os.path.exists(token_path):
        creds = Credentials.from_authorized_user_file(token_path, scopes)
    if creds is None or not creds.valid:
        if creds and creds.expired and creds.refresh_token:
            creds.refresh(Request())
        else:
            flow = InstalledAppFlow.from_client_secrets_file(credentials_path, scopes)
            creds = flow.run_local_server(port=0)
        with open(token_path, "w") as fh:
            fh.write(creds.to_json())
    # The cached token holds a long-lived OAuth refresh token; a
    # world-readable file (default umask often yields 0644) would let any
    # local user steal it.  Tighten to owner-only whenever the token file
    # exists — this also covers a still-valid token written by an older
    # version with loose permissions, where the rewrite branch above is
    # skipped entirely.
    if os.path.exists(token_path):
        try:
            os.chmod(token_path, 0o600)
        except OSError:  # pragma: no cover — e.g. unusual filesystems
            pass
    service = build("gmail", "v1", credentials=creds, cache_discovery=False)
    return cls(service)

get_history_id ¶

get_history_id() -> str

Current mailbox history cursor (users.getProfile).

One quota-cheap call that anchors incremental sync: changes after this point are retrievable via :meth:list_history_message_ids.

Source code in src/lazytools/connectors/gmail/client.py

def get_history_id(self) -> str:
    """Current mailbox history cursor (``users.getProfile``).

    One quota-cheap call that anchors incremental sync: changes after
    this point are retrievable via :meth:`list_history_message_ids`.
    """
    with self._lock:
        profile = self._service.users().getProfile(userId="me").execute()
    return str(profile["historyId"])

list_history_message_ids ¶

list_history_message_ids(*, start_history_id: str, max_results: int = 100) -> tuple[list[str], str]

Message ids added since start_history_id, plus the new cursor.

Uses users.history.list with historyTypes=messageAdded — the quota-cheap incremental alternative to re-listing the mailbox. Paginates internally (bounded), de-duplicates ids, and returns (message_ids, new_history_id); persist the returned cursor and pass it back next time.

Cursor safety. The returned cursor never advances past a message that was not returned. When the walk stops early (the max_results soft cap, or the internal page cap) while Gmail still has more history, the cursor is the id of the last fully consumed history record — so the next call resumes exactly where this one stopped instead of skipping the remainder. Only when the history was fully drained (no nextPageToken left) is Gmail's response-level historyId ("now") returned. max_results is a soft cap: a single oversized history record is always consumed whole so the caller is guaranteed to make progress.

Raises :class:GmailHistoryExpired when Gmail reports the cursor is older than its retention window (HTTP 404) — resynchronise via :meth:get_history_id.

Source code in src/lazytools/connectors/gmail/client.py

def list_history_message_ids(self, *, start_history_id: str, max_results: int = 100) -> tuple[list[str], str]:
    """Message ids added since ``start_history_id``, plus the new cursor.

    Uses ``users.history.list`` with ``historyTypes=messageAdded`` —
    the quota-cheap incremental alternative to re-listing the mailbox.
    Paginates internally (bounded), de-duplicates ids, and returns
    ``(message_ids, new_history_id)``; persist the returned cursor and
    pass it back next time.

    **Cursor safety.** The returned cursor never advances past a
    message that was not returned. When the walk stops early (the
    ``max_results`` soft cap, or the internal page cap) while Gmail
    still has more history, the cursor is the id of the **last fully
    consumed history record** — so the next call resumes exactly where
    this one stopped instead of skipping the remainder. Only when the
    history was fully drained (no ``nextPageToken`` left) is Gmail's
    response-level ``historyId`` ("now") returned. ``max_results`` is a
    soft cap: a single oversized history record is always consumed
    whole so the caller is guaranteed to make progress.

    Raises :class:`GmailHistoryExpired` when Gmail reports the cursor
    is older than its retention window (HTTP 404) — resynchronise via
    :meth:`get_history_id`.
    """
    ids: list[str] = []
    seen: set[str] = set()
    safe_cursor = str(start_history_id)  # last *fully consumed* record id
    response_cursor: str | None = None  # Gmail's "now", valid only when drained
    exhausted = False
    page_token: str | None = None
    for _ in range(20):  # hard page cap — a tick should never walk an unbounded mailbox
        with self._lock:
            request = (
                self._service.users()
                .history()
                .list(
                    userId="me",
                    startHistoryId=start_history_id,
                    historyTypes=["messageAdded"],
                    maxResults=min(max_results, 500),
                    pageToken=page_token,
                )
            )
            try:
                resp = request.execute()
            except Exception as exc:
                if _http_status(exc) == 404:
                    raise GmailHistoryExpired(
                        f"Gmail history id {start_history_id!r} has expired; resync with get_history_id()."
                    ) from exc
                raise
        response_cursor = str(resp.get("historyId", response_cursor or safe_cursor))
        stopped_mid_page = False
        for record in resp.get("history", []):
            added_ids = [a.get("message", {}).get("id") for a in record.get("messagesAdded", [])]
            fresh = [m for m in added_ids if m and m not in seen]
            # Stop *between* records once the cap is reached — but always
            # consume at least one record per call so a single oversized
            # record cannot stall the cursor forever.
            if ids and len(ids) + len(fresh) > max_results:
                stopped_mid_page = True
                break
            for message_id in fresh:
                seen.add(message_id)
                ids.append(message_id)
            record_id = record.get("id")
            if record_id is not None:
                safe_cursor = str(record_id)
        if stopped_mid_page:
            break
        page_token = resp.get("nextPageToken")
        if not page_token:
            exhausted = True
            break
        if len(ids) >= max_results:
            break
    return ids, (response_cursor if exhausted else safe_cursor) or safe_cursor

watch ¶

watch(*, topic_name: str, label_ids: list[str] | None = None) -> dict[str, Any]

Arm Gmail push notifications onto a Cloud Pub/Sub topic.

Returns the API response: {"historyId": ..., "expiration": ...} (expiration is epoch milliseconds as a string; Gmail expires a watch after at most 7 days — re-arm before then).

Source code in src/lazytools/connectors/gmail/client.py

def watch(self, *, topic_name: str, label_ids: list[str] | None = None) -> dict[str, Any]:
    """Arm Gmail push notifications onto a Cloud Pub/Sub topic.

    Returns the API response: ``{"historyId": ..., "expiration": ...}``
    (``expiration`` is epoch **milliseconds** as a string; Gmail expires
    a watch after at most 7 days — re-arm before then).
    """
    body: dict[str, Any] = {
        "topicName": topic_name,
        "labelIds": label_ids or ["INBOX"],
        "labelFilterBehavior": "INCLUDE",
    }
    with self._lock:
        return self._service.users().watch(userId="me", body=body).execute()

stop_watch ¶

stop_watch() -> None

Disarm push notifications (users.stop).

Source code in src/lazytools/connectors/gmail/client.py

def stop_watch(self) -> None:
    """Disarm push notifications (``users.stop``)."""
    with self._lock:
        self._service.users().stop(userId="me").execute()

lazytools.connectors.gmail.parse_authentication_results ¶

parse_authentication_results(header: str | None, *, trusted_authserv_id: str | None = None) -> dict[str, bool]

Return {"dkim": bool, "spf": bool, "dmarc": bool}.

True means an authoritative result token for the method was pass. A missing or empty header yields all-False.

When trusted_authserv_id is set (e.g. "mx.google.com"), the header is accepted only if its leading authserv-id is exactly that value. A forged header with a different authserv-id (or no authserv-id at all) is rejected as all-False. The match is exact rather than a prefix, so neither evil-mx.google.com nor mx.google.com.evil.com is accepted. The caller is responsible for passing the first / top-most Authentication-Results header from the message — the one prepended by the receiving MTA — rather than a later one.

Source code in src/lazytools/connectors/gmail/auth.py

def parse_authentication_results(
    header: str | None,
    *,
    trusted_authserv_id: str | None = None,
) -> dict[str, bool]:
    """Return ``{"dkim": bool, "spf": bool, "dmarc": bool}``.

    ``True`` means an authoritative result token for the method was ``pass``.
    A missing or empty header yields all-``False``.

    When ``trusted_authserv_id`` is set (e.g. ``"mx.google.com"``), the
    header is accepted **only** if its leading authserv-id is *exactly* that
    value. A forged header with a different authserv-id (or no authserv-id at
    all) is rejected as all-``False``. The match is exact rather than a
    prefix, so neither ``evil-mx.google.com`` nor ``mx.google.com.evil.com``
    is accepted. The caller is responsible for passing the first / top-most
    ``Authentication-Results`` header from the message — the one prepended by
    the receiving MTA — rather than a later one.
    """
    result = {m: False for m in _METHODS}
    if not header:
        return result

    if trusted_authserv_id is not None:
        authserv_id = _extract_authserv_id(header)
        if authserv_id != trusted_authserv_id.lower():
            return result  # authserv-id absent or does not match exactly: reject

    # Collapse comments (a few passes handles the rare nested case).
    cleaned = header
    for _ in range(5):
        stripped = _COMMENT_RE.sub(" ", cleaned)
        if stripped == cleaned:
            break
        cleaned = stripped

    for method, pattern in _RESULT_RE.items():
        # After comment-stripping + anchoring, every match is a genuine result
        # token. A message may carry several (multiple DKIM signatures); one
        # authoritative ``pass`` is enough, matching standard DKIM semantics.
        for match in pattern.finditer(cleaned):
            if match.group(1).lower() == "pass":
                result[method] = True
                break
    return result

Outlook¶

lazytools.connectors.outlook.OutlookTools ¶

OutlookTools(client: OutlookService, *, allowed_recipients: list[str] | None = None, require_confirmation: bool = True)

A ToolProvider wrapping an :class:OutlookService for the worker.

Exposes outlook_list_emails, outlook_get_email, outlook_create_draft, and outlook_send with the same Allowlist + ConfirmationGate guarding the send path as :class:GmailTools.

Source code in src/lazytools/connectors/outlook/tools.py

def __init__(
    self,
    client: OutlookService,
    *,
    allowed_recipients: list[str] | None = None,
    require_confirmation: bool = True,
) -> None:
    self._client = client
    self._allowlist = Allowlist(allowed_recipients)
    self._gate = ConfirmationGate(enabled=require_confirmation)

require_confirmation `property` ¶

require_confirmation: bool

Whether a send needs an outstanding confirmation (public attribute).

confirm_once ¶

confirm_once(*, task_id: str | None = None) -> None

Authorize exactly one send to any recipient (subject to the allow-list). Pass task_id= to bind the grant to a single task.

Source code in src/lazytools/connectors/outlook/tools.py

def confirm_once(self, *, task_id: str | None = None) -> None:
    """Authorize exactly one send to any recipient (subject to the
    allow-list). Pass ``task_id=`` to bind the grant to a single task."""
    self._gate.grant_any(scope=task_id)

confirm_send ¶

confirm_send(*, to: str, task_id: str | None = None) -> None

Authorize exactly one send to a specific recipient — the tighter, preferred grant. Pass task_id= to also bind it to a single task.

Source code in src/lazytools/connectors/outlook/tools.py

def confirm_send(self, *, to: str, task_id: str | None = None) -> None:
    """Authorize exactly one send to a specific recipient — the tighter,
    preferred grant. Pass ``task_id=`` to also bind it to a single task."""
    self._gate.grant(to, scope=task_id)

lazytools.connectors.outlook.OutlookClient ¶

OutlookClient(namespace: Any, application: Any, *, folder_index: int = _OL_FOLDER_INBOX, executor: Executor | None = None)

Production :class:OutlookService backed by a local Outlook via COM.

Build one with :meth:connect, which attaches to the running/registered Outlook application on a dedicated single worker thread that has called :func:pythoncom.CoInitialize. Every COM access — the initial Dispatch and all reads/sends — is marshalled onto that one thread, so the client is safe to call from any thread (including asyncio.to_thread executor threads) without apartment/initialisation errors. Call :meth:close to shut the worker down.

Source code in src/lazytools/connectors/outlook/client.py

def __init__(
    self,
    namespace: Any,
    application: Any,
    *,
    folder_index: int = _OL_FOLDER_INBOX,
    executor: concurrent.futures.Executor | None = None,
) -> None:
    # ``namespace`` is a MAPI namespace; ``application`` is the Outlook
    # Application object (kept for CreateItem on the send path).
    #
    # ``executor`` is the single-thread, CoInitialized COM worker. When
    # ``None`` (direct injection, e.g. tests with fakes) calls run inline
    # on the caller's thread — fakes have no thread affinity.
    self._namespace = namespace
    self._application = application
    self._folder_index = folder_index
    self._executor = executor
    self._lock = threading.Lock()

connect `classmethod` ¶

connect(*, folder_index: int = _OL_FOLDER_INBOX) -> OutlookClient

Attach to the local Outlook desktop application via COM.

Imports pywin32 lazily; raises a friendly ImportError if the outlook extra (or a non-Windows platform) makes it unavailable. Outlook must be installed and signed in; the call reuses that session, so there are no separate credentials to manage.

The Dispatch runs on the dedicated CoInitialized worker thread that will own every subsequent COM call — so the proxy is created and used on one apartment-correct thread.

Source code in src/lazytools/connectors/outlook/client.py

@classmethod
def connect(cls, *, folder_index: int = _OL_FOLDER_INBOX) -> OutlookClient:
    """Attach to the local Outlook desktop application via COM.

    Imports ``pywin32`` lazily; raises a friendly ``ImportError`` if the
    ``outlook`` extra (or a non-Windows platform) makes it unavailable.
    Outlook must be installed and signed in; the call reuses that session,
    so there are no separate credentials to manage.

    The ``Dispatch`` runs on the dedicated CoInitialized worker thread that
    will own every subsequent COM call — so the proxy is created and used
    on one apartment-correct thread.
    """
    try:
        import win32com.client  # type: ignore[import-not-found]  # noqa: F401
    except ImportError as exc:  # pragma: no cover — exercised only without the extra
        raise ImportError(
            "OutlookClient.connect requires the 'outlook' extra on Windows "
            "(local Outlook desktop + pywin32). Install it with: "
            'pip install "lazytoolkit[outlook] @ git+https://github.com/selvaz/LazyTools.git"'
        ) from exc

    executor = concurrent.futures.ThreadPoolExecutor(
        max_workers=1, thread_name_prefix="outlook-com", initializer=_init_com
    )

    def _dispatch() -> tuple[Any, Any]:  # pragma: no cover — needs real COM
        import win32com.client as _w

        application = _w.Dispatch("Outlook.Application")
        namespace = application.GetNamespace("MAPI")
        return application, namespace

    application, namespace = executor.submit(_dispatch).result()
    return cls(namespace, application, folder_index=folder_index, executor=executor)

close ¶

close() -> None

Shut down the COM worker thread (no-op for injected clients).

Source code in src/lazytools/connectors/outlook/client.py

def close(self) -> None:
    """Shut down the COM worker thread (no-op for injected clients)."""
    if self._executor is not None:
        self._executor.shutdown(wait=True)
        self._executor = None

list_message_ids ¶

list_message_ids(*, query: str | None = None, max_results: int = 25) -> list[str]

Entry IDs of messages in the watched folder, newest first.

query is an Outlook Restrict filter (DASL or the "[Field] = 'value'" macro syntax), e.g. "[Unread] = true"; None returns the whole folder (capped at max_results).

Source code in src/lazytools/connectors/outlook/client.py

def list_message_ids(self, *, query: str | None = None, max_results: int = 25) -> list[str]:
    """Entry IDs of messages in the watched folder, newest first.

    ``query`` is an Outlook **Restrict** filter (DASL or the
    ``"[Field] = 'value'"`` macro syntax), e.g. ``"[Unread] = true"``;
    ``None`` returns the whole folder (capped at ``max_results``).
    """

    def _op() -> list[str]:
        with self._lock:
            folder = self._namespace.GetDefaultFolder(self._folder_index)
            items = folder.Items
            items.Sort("[ReceivedTime]", True)  # descending → newest first
            if query:
                items = items.Restrict(query)
            out: list[str] = []
            item = items.GetFirst()
            while item is not None and len(out) < max_results:
                entry_id = getattr(item, "EntryID", None)
                if entry_id:
                    out.append(entry_id)
                item = items.GetNext()
            return out

    return self._run(_op)

Telegram¶

lazytools.connectors.telegram.TelegramTools ¶

TelegramTools(client: TelegramService, *, allowed_chat_ids: list[int | str] | None = None, require_confirmation: bool = True, attachments_dir: str | PathLike[str] | None = None)

A ToolProvider wrapping a :class:TelegramService for the worker.

Source code in src/lazytools/connectors/telegram/tools.py

def __init__(
    self,
    client: TelegramService,
    *,
    allowed_chat_ids: list[int | str] | None = None,
    require_confirmation: bool = True,
    attachments_dir: str | os.PathLike[str] | None = None,
) -> None:
    self._client = client
    self._allowlist = Allowlist(allowed_chat_ids)
    self._gate = ConfirmationGate(enabled=require_confirmation)
    # When set, ``telegram_send_document`` may only upload files resolving
    # *under* this directory. ``file_path`` is typically model-controlled,
    # so confining it stops an agent from exfiltrating arbitrary host files.
    # ``None`` permits any path (trusted-caller mode) — mirrors how
    # ``allowed_chat_ids=None`` permits any chat. Set it whenever the tool
    # is exposed to an LLM (e.g. to the directory ``save_report`` writes to).
    self._attachments_dir = os.path.realpath(os.fspath(attachments_dir)) if attachments_dir is not None else None
    # The single-owner-bot deployment (the docstring's "reply freely to a
    # known chat" case) has exactly one possible destination -- making an
    # LLM supply a chat_id argument that can only ever be one value is
    # just a chance for it to get it wrong instead of reusing the one
    # it's already talking in (observed in practice: a cheaper model
    # drifted to a different id after enough turns and every send then
    # failed the allow-list). When allowed_chat_ids has exactly one
    # entry, chat_id becomes optional on the outbound tools and defaults
    # to it; multi-chat or unrestricted deployments still require it
    # explicitly.
    ids = list(allowed_chat_ids) if allowed_chat_ids is not None else []
    self._default_chat_id: int | str | None = ids[0] if len(ids) == 1 else None

require_confirmation `property` ¶

require_confirmation: bool

Whether a send needs an outstanding confirmation (public attribute).

confirm_once ¶

confirm_once(*, task_id: str | None = None) -> None

Authorize exactly one send to any chat (subject to the allow-list). Pass task_id= to bind the grant to a single task so a concurrent task cannot consume it.

Source code in src/lazytools/connectors/telegram/tools.py

def confirm_once(self, *, task_id: str | None = None) -> None:
    """Authorize exactly one send to any chat (subject to the allow-list).
    Pass ``task_id=`` to bind the grant to a single task so a concurrent
    task cannot consume it."""
    self._gate.grant_any(scope=task_id)

confirm_send ¶

confirm_send(*, chat_id: int | str, task_id: str | None = None) -> None

Authorize exactly one send to a specific chat — the tighter grant. Pass task_id= to also bind it to a single task.

Source code in src/lazytools/connectors/telegram/tools.py

def confirm_send(self, *, chat_id: int | str, task_id: str | None = None) -> None:
    """Authorize exactly one send to a specific chat — the tighter grant.
    Pass ``task_id=`` to also bind it to a single task."""
    self._gate.grant(chat_id, scope=task_id)

lazytools.connectors.telegram.TelegramClient ¶

TelegramClient(token: str, *, http: Any | None = None, base_url: str = 'https://api.telegram.org')

Production :class:TelegramService backed by the Bot API over HTTPS.

Source code in src/lazytools/connectors/telegram/client.py

def __init__(self, token: str, *, http: Any | None = None, base_url: str = "https://api.telegram.org") -> None:
    # ``http`` is an ``httpx.Client`` (or any object exposing
    # ``post(url, json=...) -> response`` with ``raise_for_status`` + ``json``).
    self._token = token
    self._base = f"{base_url}/bot{token}"
    self._http = http

from_token `classmethod` ¶

from_token(token: str, *, timeout: float = 30.0) -> TelegramClient

Build a client from a bot token (obtained from @BotFather).

Imports httpx lazily; raises a friendly ImportError if the telegram extra is not installed.

Source code in src/lazytools/connectors/telegram/client.py

@classmethod
def from_token(cls, token: str, *, timeout: float = 30.0) -> TelegramClient:
    """Build a client from a bot token (obtained from @BotFather).

    Imports ``httpx`` lazily; raises a friendly ``ImportError`` if the
    ``telegram`` extra is not installed.
    """
    try:
        import httpx
    except ImportError as exc:  # pragma: no cover — exercised only without the extra
        raise ImportError(
            "TelegramClient.from_token requires the 'telegram' extra. "
            'Install it with: pip install "lazytoolkit[telegram] @ git+https://github.com/selvaz/LazyTools.git"'
        ) from exc
    return cls(token, http=httpx.Client(timeout=timeout))

close ¶

close() -> None

Close the underlying HTTP client (its connection pool), if it has one.

Source code in src/lazytools/connectors/telegram/client.py

def close(self) -> None:
    """Close the underlying HTTP client (its connection pool), if it has one."""
    close = getattr(self._http, "close", None)
    if callable(close):
        close()

send_document ¶

send_document(*, chat_id: int | str, document: bytes, filename: str = 'document', caption: str | None = None) -> dict[str, Any]

Upload document (raw bytes) to chat_id via sendDocument.

filename is the name shown in Telegram; caption is optional (≤1024 chars, enforced by the caller). Returns the Bot API result.

Source code in src/lazytools/connectors/telegram/client.py

def send_document(
    self,
    *,
    chat_id: int | str,
    document: bytes,
    filename: str = "document",
    caption: str | None = None,
) -> dict[str, Any]:
    """Upload ``document`` (raw bytes) to ``chat_id`` via ``sendDocument``.

    ``filename`` is the name shown in Telegram; ``caption`` is optional
    (≤1024 chars, enforced by the caller). Returns the Bot API ``result``.
    """
    data: dict[str, Any] = {"chat_id": chat_id}
    if caption:
        data["caption"] = caption
    files = {"document": (filename, document)}
    return dict(self._call_multipart("sendDocument", data, files) or {})

MCP¶

lazytools.connectors.mcp.MCP ¶

Public factory for :class:MCPServer instances.

stdio `classmethod` ¶

stdio(name: str, *, command: str, args: list[str] | None = None, env: dict[str, str] | None = None, namespace: bool = True, prefix: str | None = None, allow: Iterable[str] | None = None, deny: Iterable[str] | None = None, cache_tools_ttl: float | None = MCPServer._DEFAULT_CACHE_TTL) -> MCPServer

Build an MCP server bound to a stdio (subprocess) transport.

allow= (or deny=) is required — same deny-by-default posture as :meth:http. A warn-and-proceed default would expose every advertised tool to the LLM silently; that's a non-trivial blast radius for filesystem / git / shell MCP servers, so we fail at construction instead. Pass allow=["*"] to opt every advertised tool in explicitly after auditing the surface.

Source code in src/lazytools/connectors/mcp/server.py

@classmethod
def stdio(
    cls,
    name: str,
    *,
    command: str,
    args: list[str] | None = None,
    env: dict[str, str] | None = None,
    namespace: bool = True,
    prefix: str | None = None,
    allow: Iterable[str] | None = None,
    deny: Iterable[str] | None = None,
    cache_tools_ttl: float | None = MCPServer._DEFAULT_CACHE_TTL,
) -> MCPServer:
    """Build an MCP server bound to a stdio (subprocess) transport.

    ``allow=`` (or ``deny=``) is **required** — same deny-by-default
    posture as :meth:`http`.  A warn-and-proceed default would expose
    every advertised tool to the LLM silently; that's a non-trivial
    blast radius for filesystem / git / shell MCP servers, so we fail
    at construction instead.  Pass ``allow=["*"]`` to opt every
    advertised tool in explicitly after auditing the surface.
    """
    if allow is None and deny is None:
        raise ValueError(
            f"MCP.stdio({name!r}, command={command!r}) requires an explicit\n"
            f"  allow=[...] or deny=[...] filter (deny-by-default).\n"
            f"  Pass allow=['*'] to opt every advertised tool in after auditing\n"
            f"  the surface, or pass an explicit allow / deny list of fnmatch\n"
            f"  globs (e.g. allow=['fs.read_*', 'fs.list_*']).\n"
            f"  A warn-and-proceed default would be unsafe for filesystem /\n"
            f"  git / shell MCP servers — the LLM could invoke any tool the\n"
            f"  subprocess advertised."
        )
    from lazytools.connectors.mcp.transports import StdioTransport

    return MCPServer(
        name,
        transport=StdioTransport(command, args=args, env=env),
        namespace=namespace,
        prefix=prefix,
        allow=allow,
        deny=deny,
        cache_tools_ttl=cache_tools_ttl,
    )

http `classmethod` ¶

http(name: str, url: str, *, headers: dict[str, str] | None = None, namespace: bool = True, prefix: str | None = None, allow: Iterable[str] | None = None, deny: Iterable[str] | None = None, cache_tools_ttl: float | None = MCPServer._DEFAULT_CACHE_TTL) -> MCPServer

Build an MCP server bound to a Streamable HTTP transport.

allow= is required. Omitting it raises ValueError because a remote server could advertise any number of tools and silently exposing them all to the LLM is a security mistake. Pass an explicit list of the tools you want to expose::

MCP.http("github", url, allow=["create_issue", "list_prs"])

To permit all tools advertised by a server you fully control::

MCP.http("internal", url, allow=["*"])

Source code in src/lazytools/connectors/mcp/server.py

@classmethod
def http(
    cls,
    name: str,
    url: str,
    *,
    headers: dict[str, str] | None = None,
    namespace: bool = True,
    prefix: str | None = None,
    allow: Iterable[str] | None = None,
    deny: Iterable[str] | None = None,
    cache_tools_ttl: float | None = MCPServer._DEFAULT_CACHE_TTL,
) -> MCPServer:
    """Build an MCP server bound to a Streamable HTTP transport.

    ``allow=`` is **required**. Omitting it raises ``ValueError`` because
    a remote server could advertise any number of tools and silently
    exposing them all to the LLM is a security mistake. Pass an explicit
    list of the tools you want to expose::

        MCP.http("github", url, allow=["create_issue", "list_prs"])

    To permit all tools advertised by a server you fully control::

        MCP.http("internal", url, allow=["*"])
    """
    if allow is None:
        raise ValueError(
            f"MCP.http({name!r}, {url!r}) requires an explicit allow= list. "
            f"Every tool the remote server advertises would otherwise be exposed to the LLM. "
            f"Pass allow=['tool_a', 'tool_b'] to restrict the tool surface, "
            f"or allow=['*'] to permit everything and silence this error."
        )
    from lazytools.connectors.mcp.transports import HttpTransport

    return MCPServer(
        name,
        transport=HttpTransport(url, headers=headers),
        namespace=namespace,
        prefix=prefix,
        allow=allow,
        deny=deny,
        cache_tools_ttl=cache_tools_ttl,
    )

from_transport `classmethod` ¶

from_transport(name: str, transport: _Transport, *, namespace: bool = True, prefix: str | None = None, allow: Iterable[str] | None = None, deny: Iterable[str] | None = None, cache_tools_ttl: float | None = MCPServer._DEFAULT_CACHE_TTL) -> MCPServer

Build an MCP server from a custom :class:_Transport.

Useful for tests (in-process fake transport) or for adapters to non-standard MCP variants. The transport must implement the abstract :class:_Transport interface.

Loop contract. All transport methods run on the server's dedicated background loop, never on the caller's loop — this is what keeps loop-affine sessions alive across the sync as_tools() facade and arbitrary caller loops. Custom transports must create loop-affine resources lazily inside connect() rather than binding them to the caller's loop beforehand. (Pre-bound resources never worked reliably: the sync facade previously ran them on a throwaway asyncio.run loop.)

Source code in src/lazytools/connectors/mcp/server.py

@classmethod
def from_transport(
    cls,
    name: str,
    transport: _Transport,
    *,
    namespace: bool = True,
    prefix: str | None = None,
    allow: Iterable[str] | None = None,
    deny: Iterable[str] | None = None,
    cache_tools_ttl: float | None = MCPServer._DEFAULT_CACHE_TTL,
) -> MCPServer:
    """Build an MCP server from a custom :class:`_Transport`.

    Useful for tests (in-process fake transport) or for adapters to
    non-standard MCP variants. The transport must implement the
    abstract :class:`_Transport` interface.

    **Loop contract.** All transport methods run on the server's
    dedicated background loop, never on the caller's loop — this is
    what keeps loop-affine sessions alive across the sync
    ``as_tools()`` facade and arbitrary caller loops. Custom
    transports must create loop-affine resources lazily inside
    ``connect()`` rather than binding them to the caller's loop
    beforehand. (Pre-bound resources never worked reliably: the sync
    facade previously ran them on a throwaway ``asyncio.run`` loop.)
    """
    return MCPServer(
        name,
        transport=transport,
        namespace=namespace,
        prefix=prefix,
        allow=allow,
        deny=deny,
        cache_tools_ttl=cache_tools_ttl,
    )

lazytools.connectors.mcp.MCPServer ¶

MCPServer(name: str, transport: _Transport, *, namespace: bool = True, prefix: str | None = None, allow: Iterable[str] | None = None, deny: Iterable[str] | None = None, cache_tools_ttl: float | None = _DEFAULT_CACHE_TTL)

A tool provider backed by an MCP server.

Add it directly to Agent(tools=[...]); the framework calls :meth:as_tools to expand it into individual :class:Tool entries. Tool names are namespaced as "<server-name>.<mcp-tool-name>" by default; pass namespace=False to keep the raw names, or prefix="..." to override.

The transport connects lazily on first :meth:as_tools. For explicit cleanup, use the server as an async context manager::

async with MCP.stdio("fs", command="...", args=[...]) as fs:
    agent = Agent("claude-opus-4-8", tools=[fs])
    await agent.run("...")

Without that, the transport stays open for the process lifetime; the underlying subprocess is normally cleaned up when the parent exits.

Closure is terminal. Once :meth:aclose (or the async with block) finishes, the server is single-shot: a subsequent :meth:aconnect / :meth:as_tools raises RuntimeError. Construct a new MCPServer if you need to re-use the same transport configuration.

Source code in src/lazytools/connectors/mcp/server.py

def __init__(
    self,
    name: str,
    transport: _Transport,
    *,
    namespace: bool = True,
    prefix: str | None = None,
    allow: Iterable[str] | None = None,
    deny: Iterable[str] | None = None,
    cache_tools_ttl: float | None = _DEFAULT_CACHE_TTL,
) -> None:
    self.name = name
    self._transport = transport
    self._namespace = namespace
    if prefix is not None:
        self._prefix = prefix
    else:
        self._prefix = f"{name}." if namespace else ""
    self._allow = list(allow) if allow else None
    self._deny = list(deny) if deny else None

    if cache_tools_ttl is not None and cache_tools_ttl <= 0:
        raise ValueError(f"cache_tools_ttl must be > 0 or None, got {cache_tools_ttl!r}")
    self._cache_ttl: float | None = cache_tools_ttl
    self._tools_cache: list[Tool] | None = None
    self._tools_cache_ts: float = 0.0
    self._connected = False
    self._closed = False
    # Lazy-init the asyncio.Lock on first async use.  It is created and
    # only ever acquired on the runner loop (see ``_get_runner``), so it
    # can never be bound to a caller's loop.
    self._lock: asyncio.Lock | None = None
    # Dedicated background loop that owns the transport session.  The
    # official SDK's sessions are loop-affine: created lazily on first
    # use, every transport operation is dispatched onto this loop so the
    # session is created, used, and closed on one loop regardless of
    # which loop (or none) the caller is on.
    self._runner: _LoopRunner | None = None
    self._runner_guard = threading.Lock()

aconnect `async` ¶

aconnect() -> None

Connect the underlying transport. Idempotent.

Source code in src/lazytools/connectors/mcp/server.py

async def aconnect(self) -> None:
    """Connect the underlying transport. Idempotent."""
    await self._get_runner().run(self._aconnect_impl())

alist_tools `async` ¶

alist_tools() -> list[Tool]

Discover and wrap the server's tools.

Cached for cache_tools_ttl seconds (default 60 s). Once the cache expires the next call re-fetches from the upstream transport so an MCP server that hot-loads or unloads tools is eventually reflected in the agent's tool list. Pass cache_tools_ttl=None to disable expiry entirely and :meth:invalidate_tools_cache to flush explicitly.

Returns a fresh list on every call; mutating it does not affect the server's internal cache.

Source code in src/lazytools/connectors/mcp/server.py

async def alist_tools(self) -> list[Tool]:
    """Discover and wrap the server's tools.

    Cached for ``cache_tools_ttl`` seconds (default 60 s).  Once the
    cache expires the next call re-fetches from the upstream
    transport so an MCP server that hot-loads or unloads tools is
    eventually reflected in the agent's tool list.
    Pass ``cache_tools_ttl=None`` to disable expiry entirely and
    :meth:`invalidate_tools_cache` to flush explicitly.

    Returns a fresh list on every call; mutating it does not affect
    the server's internal cache.
    """
    return await self._get_runner().run(self._alist_tools_impl())

invalidate_tools_cache ¶

invalidate_tools_cache() -> None

Drop the cached tool list so the next call re-fetches.

Use this when an out-of-band signal tells you the MCP server's tool registry has changed (plugin install / uninstall, hot reload). No-op when nothing is cached yet.

Source code in src/lazytools/connectors/mcp/server.py

def invalidate_tools_cache(self) -> None:
    """Drop the cached tool list so the next call re-fetches.

    Use this when an out-of-band signal tells you the MCP server's
    tool registry has changed (plugin install / uninstall, hot
    reload).  No-op when nothing is cached yet.
    """
    self._tools_cache = None
    self._tools_cache_ts = 0.0

aclose `async` ¶

aclose() -> None

Close the underlying transport and stop the background loop.

Idempotent, and terminal even when the server was never connected: any later :meth:aconnect / :meth:as_tools raises RuntimeError.

Source code in src/lazytools/connectors/mcp/server.py

async def aclose(self) -> None:
    """Close the underlying transport and stop the background loop.

    Idempotent, and terminal even when the server was never connected:
    any later :meth:`aconnect` / :meth:`as_tools` raises ``RuntimeError``.
    """
    with self._runner_guard:
        runner = self._runner
        self._runner = None
        self._closed = True
    if runner is None:
        return
    try:
        await runner.run(self._aclose_impl())
    finally:
        await asyncio.to_thread(runner.stop)

as_tools ¶

as_tools() -> list[Tool]

Sync wrapper around :meth:alist_tools. Called by build_tool_map.

Triggers a lazy connect on first use. The work always runs on the server's dedicated background loop, so it is safe to call this with or without a running event loop — the session survives for later tool calls either way (it is never bound to a throwaway loop).

Source code in src/lazytools/connectors/mcp/server.py

def as_tools(self) -> list[Tool]:
    """Sync wrapper around :meth:`alist_tools`. Called by ``build_tool_map``.

    Triggers a lazy connect on first use. The work always runs on the
    server's dedicated background loop, so it is safe to call this with
    or without a running event loop — the session survives for later
    tool calls either way (it is never bound to a throwaway loop).
    """
    return self._get_runner().run_sync(self._alist_tools_impl())

Code Support Agent¶

lazytools.connectors.code_support.claude_code ¶

claude_code(task: str, *, mode: str = 'read', cwd: str | None = None, session_id: str | None = None, timeout: float = 300.0, model: str | None = 'claude-sonnet-5') -> dict[str, Any] | str

Delegate a read-only task to Claude Code CLI.

Returns {"result": <text>, "content_is_untrusted": true} on success — the result is derived from whatever code/text the CLI read, so downstream consumers must treat it as third-party content (the same labelling convention as the EDGAR connector). Connector-level failures (CLI missing, timeout, non-zero exit) return a plain "[claude_code] ..." string.

Parameters¶

task: Instruction for Claude Code. mode: "read" (default) — read-only analysis (Read, Grep, Glob; no Bash, so the CLI cannot run commands or modify files). "plan" — plan mode, no file modifications.

There is deliberately no ``"write"`` here: file edits and command
execution live behind
:class:`~lazytools.connectors.code_support.CodeWriteTools`, which
requires an explicit ``base_dir`` sandbox and (by default) a one-shot
confirmation per write call — so an orchestrating LLM can only write
if the developer handed it the writer tool.

cwd: Working directory for the subprocess. Note that read mode can read anything the process user can read — point cwd at the project, and prefer running the whole agent under a low-privilege user if the machine holds secrets. session_id: If given, resumes an existing Claude Code session via --resume. timeout: Maximum seconds for the subprocess. Set tool_timeout=None on LLMEngine so the engine never cancels before the subprocess finishes (zombie-process hazard when engine fires first). model: --model passed to the CLI. Defaults to "claude-sonnet-5" so the delegated session has a pinned, predictable model regardless of the CLI's own interactive default; pass an alias ("opus", "sonnet") or a full model name, or None to omit the flag and let the CLI decide.

Notes¶

Auth is left to the Claude Code CLI itself: it reads its own on-disk login (~/.claude/.credentials.json), and the subprocess inherits the current environment, so CLAUDE_CODE_OAUTH_TOKEN (from claude setup-token) or ANTHROPIC_API_KEY are honoured if set. We do not synthesize CLAUDE_CODE_OAUTH_TOKEN from the credentials file — that env var is a token string, not the JSON store, and overriding it would break a valid disk login.

Source code in src/lazytools/connectors/code_support/_claude_code.py

def claude_code(
    task: str,
    *,
    mode: str = "read",
    cwd: str | None = None,
    session_id: str | None = None,
    timeout: float = 300.0,
    model: str | None = "claude-sonnet-5",
) -> dict[str, Any] | str:
    """Delegate a read-only task to Claude Code CLI.

    Returns ``{"result": <text>, "content_is_untrusted": true}`` on success —
    the result is derived from whatever code/text the CLI read, so downstream
    consumers must treat it as third-party content (the same labelling
    convention as the EDGAR connector). Connector-level failures (CLI missing,
    timeout, non-zero exit) return a plain ``"[claude_code] ..."`` string.

    Parameters
    ----------
    task:
        Instruction for Claude Code.
    mode:
        ``"read"`` (default) — read-only analysis (Read, Grep, Glob; no
        Bash, so the CLI cannot run commands or modify files).
        ``"plan"`` — plan mode, no file modifications.

        There is deliberately no ``"write"`` here: file edits and command
        execution live behind
        :class:`~lazytools.connectors.code_support.CodeWriteTools`, which
        requires an explicit ``base_dir`` sandbox and (by default) a one-shot
        confirmation per write call — so an orchestrating LLM can only write
        if the developer handed it the writer tool.
    cwd:
        Working directory for the subprocess. Note that read mode can read
        anything the process user can read — point ``cwd`` at the project,
        and prefer running the whole agent under a low-privilege user if the
        machine holds secrets.
    session_id:
        If given, resumes an existing Claude Code session via ``--resume``.
    timeout:
        Maximum seconds for the subprocess. Set ``tool_timeout=None`` on
        ``LLMEngine`` so the engine never cancels before the subprocess
        finishes (zombie-process hazard when engine fires first).
    model:
        ``--model`` passed to the CLI. Defaults to ``"claude-sonnet-5"`` so
        the delegated session has a pinned, predictable model regardless of
        the CLI's own interactive default; pass an alias (``"opus"``,
        ``"sonnet"``) or a full model name, or ``None`` to omit the flag and
        let the CLI decide.

    Notes
    -----
    Auth is left to the Claude Code CLI itself: it reads its own on-disk
    login (``~/.claude/.credentials.json``), and the subprocess inherits the
    current environment, so ``CLAUDE_CODE_OAUTH_TOKEN`` (from
    ``claude setup-token``) or ``ANTHROPIC_API_KEY`` are honoured if set. We do
    not synthesize ``CLAUDE_CODE_OAUTH_TOKEN`` from the credentials file — that
    env var is a token *string*, not the JSON store, and overriding it would
    break a valid disk login.
    """
    if mode not in _TOOL_FLAGS:
        return (
            f"[claude_code] invalid mode={mode!r}. Use 'read' or 'plan'. "
            "Writes require the gated CodeWriteTools provider "
            "(lazytools.connectors.code_support.CodeWriteTools)."
        )

    out = _run_claude(
        task, _TOOL_FLAGS[mode], cwd=cwd, session_id=session_id, timeout=timeout, model=model
    )
    if out.startswith("[claude_code]"):
        return out
    return {"result": out, "content_is_untrusted": True}

lazytools.connectors.code_support.claude_code_mcp ¶

claude_code_mcp(*, name: str = 'claude_code', allow: Iterable[str] | None = None, deny: Iterable[str] | None = None, args: list[str] | None = None, env: dict[str, str] | None = None, namespace: bool = True, prefix: str | None = None, cache_tools_ttl: float | None = 60.0) -> MCPServer

Claude Code as an MCP server (claude mcp serve).

Returns an :class:~lazytools.connectors.mcp.MCPServer exposing Claude Code's own tools (View, Edit, LS, Bash, …) over stdio. Drop it straight into Agent(tools=[claude_code_mcp(allow=["*"])]).

allow= (or deny=) is required — deny-by-default, the same posture as :meth:MCP.stdio. The patterns match the namespaced tool names, e.g. allow=["claude_code.View", "claude_code.LS"] (or allow=["*"] after auditing the surface). Tool names are not hardcoded here because they are owned by the Claude Code version you have installed; discover them by running with allow=["*"] once and inspecting the map.

Parameters¶

name: Server name and default namespace prefix ("claude_code"). allow / deny: fnmatch globs against the namespaced tool name (deny-by-default). args: Extra args appended after mcp serve (rarely needed). env: Extra environment for the subprocess. Auth is otherwise inherited from the parent environment / the CLI's own on-disk login. namespace / prefix / cache_tools_ttl: Forwarded to :meth:MCP.stdio unchanged.

Requires the mcp extra: pip install "lazytoolkit[mcp] @ git+https://github.com/selvaz/LazyTools.git".

Source code in src/lazytools/connectors/code_support/_claude_code.py

def claude_code_mcp(
    *,
    name: str = "claude_code",
    allow: Iterable[str] | None = None,
    deny: Iterable[str] | None = None,
    args: list[str] | None = None,
    env: dict[str, str] | None = None,
    namespace: bool = True,
    prefix: str | None = None,
    cache_tools_ttl: float | None = 60.0,
) -> MCPServer:
    """Claude Code as an MCP server (``claude mcp serve``).

    Returns an :class:`~lazytools.connectors.mcp.MCPServer` exposing Claude
    Code's own tools (View, Edit, LS, Bash, …) over stdio. Drop it straight into
    ``Agent(tools=[claude_code_mcp(allow=["*"])])``.

    ``allow=`` (or ``deny=``) is **required** — deny-by-default, the same
    posture as :meth:`MCP.stdio`. The patterns match the *namespaced* tool
    names, e.g. ``allow=["claude_code.View", "claude_code.LS"]`` (or
    ``allow=["*"]`` after auditing the surface). Tool names are not hardcoded
    here because they are owned by the Claude Code version you have installed;
    discover them by running with ``allow=["*"]`` once and inspecting the map.

    Parameters
    ----------
    name:
        Server name and default namespace prefix (``"claude_code"``).
    allow / deny:
        fnmatch globs against the namespaced tool name (deny-by-default).
    args:
        Extra args appended after ``mcp serve`` (rarely needed).
    env:
        Extra environment for the subprocess. Auth is otherwise inherited
        from the parent environment / the CLI's own on-disk login.
    namespace / prefix / cache_tools_ttl:
        Forwarded to :meth:`MCP.stdio` unchanged.

    Requires the ``mcp`` extra: ``pip install "lazytoolkit[mcp] @ git+https://github.com/selvaz/LazyTools.git"``.
    """
    return MCP.stdio(
        name,
        command="claude",
        args=["mcp", "serve", *(args or [])],
        env=env,
        allow=allow,
        deny=deny,
        namespace=namespace,
        prefix=prefix,
        cache_tools_ttl=cache_tools_ttl,
    )

lazytools.connectors.code_support.codex ¶

codex(task: str, *, cwd: str | None = None, resume_last: bool = False, timeout: float = 300.0, skip_git_check: bool = True) -> dict[str, Any] | str

Delegate a read-only task to the Codex CLI (-s read-only sandbox).

Returns {"result": <text>, "content_is_untrusted": true} on success — the result is derived from whatever code/text the CLI read, so downstream consumers must treat it as third-party content. Connector-level failures (CLI missing, timeout, non-zero exit) return a plain "[codex] ..." string.

There is deliberately no write mode here: file edits live behind :class:~lazytools.connectors.code_support.CodeWriteTools, which requires an explicit base_dir sandbox and (by default) a one-shot confirmation per write call.

Parameters¶

task: Instruction for Codex. cwd: Working directory for the subprocess. Read-only sandbox still reads anything the process user can read — point cwd at the project and prefer a low-privilege user on machines that hold secrets. resume_last: If True, continues the most recent Codex session in the working directory via exec resume --last. timeout: Maximum seconds for the subprocess. Set tool_timeout=None on LLMEngine so the engine never cancels before the subprocess finishes (zombie-process hazard when engine fires first). skip_git_check: Pass --skip-git-repo-check. Harmless in the read-only sandbox; the gated writer defaults this off so writes keep git as a recovery rail.

Source code in src/lazytools/connectors/code_support/_codex.py

def codex(
    task: str,
    *,
    cwd: str | None = None,
    resume_last: bool = False,
    timeout: float = 300.0,
    skip_git_check: bool = True,
) -> dict[str, Any] | str:
    """Delegate a read-only task to the Codex CLI (``-s read-only`` sandbox).

    Returns ``{"result": <text>, "content_is_untrusted": true}`` on success —
    the result is derived from whatever code/text the CLI read, so downstream
    consumers must treat it as third-party content. Connector-level failures
    (CLI missing, timeout, non-zero exit) return a plain ``"[codex] ..."``
    string.

    There is deliberately no write mode here: file edits live behind
    :class:`~lazytools.connectors.code_support.CodeWriteTools`, which requires
    an explicit ``base_dir`` sandbox and (by default) a one-shot confirmation
    per write call.

    Parameters
    ----------
    task:
        Instruction for Codex.
    cwd:
        Working directory for the subprocess. Read-only sandbox still reads
        anything the process user can read — point ``cwd`` at the project and
        prefer a low-privilege user on machines that hold secrets.
    resume_last:
        If True, continues the most recent Codex session in the working
        directory via ``exec resume --last``.
    timeout:
        Maximum seconds for the subprocess. Set ``tool_timeout=None`` on
        ``LLMEngine`` so the engine never cancels before the subprocess
        finishes (zombie-process hazard when engine fires first).
    skip_git_check:
        Pass ``--skip-git-repo-check``. Harmless in the read-only sandbox;
        the gated writer defaults this **off** so writes keep git as a
        recovery rail.
    """
    out = _run_codex(
        task,
        _READ_FLAGS,
        cwd=cwd,
        resume_last=resume_last,
        skip_git_check=skip_git_check,
        timeout=timeout,
    )
    if out.startswith("[codex]"):
        return out
    return {"result": out, "content_is_untrusted": True}

lazytools.connectors.code_support.codex_mcp ¶

codex_mcp(*, name: str = 'codex', allow: Iterable[str] | None = None, deny: Iterable[str] | None = None, args: list[str] | None = None, env: dict[str, str] | None = None, namespace: bool = True, prefix: str | None = None, cache_tools_ttl: float | None = 60.0) -> MCPServer

Codex as an MCP server (codex mcp-server).

Returns an :class:~lazytools.connectors.mcp.MCPServer exposing Codex's MCP interface over stdio. Drop it into Agent(tools=[codex_mcp(allow=["*"])]).

Warning

Codex's MCP-server interface is experimental (per OpenAI's docs) and may change without notice. Pin your Codex version if you depend on the exposed tool shape.

allow= (or deny=) is required — deny-by-default. Patterns match the namespaced tool name (codex.*). Tool names are not hardcoded because they depend on the installed Codex version; discover them with allow=["*"] once.

Parameters¶

name: Server name and default namespace prefix ("codex"). allow / deny: fnmatch globs against the namespaced tool name (deny-by-default). args: Extra args appended after mcp-server (rarely needed). env: Extra environment for the subprocess. Auth is otherwise inherited from codex login / the current shell environment. namespace / prefix / cache_tools_ttl: Forwarded to :meth:MCP.stdio unchanged.

Requires the mcp extra: pip install "lazytoolkit[mcp] @ git+https://github.com/selvaz/LazyTools.git".

Source code in src/lazytools/connectors/code_support/_codex.py

def codex_mcp(
    *,
    name: str = "codex",
    allow: Iterable[str] | None = None,
    deny: Iterable[str] | None = None,
    args: list[str] | None = None,
    env: dict[str, str] | None = None,
    namespace: bool = True,
    prefix: str | None = None,
    cache_tools_ttl: float | None = 60.0,
) -> MCPServer:
    """Codex as an MCP server (``codex mcp-server``).

    Returns an :class:`~lazytools.connectors.mcp.MCPServer` exposing Codex's
    MCP interface over stdio. Drop it into
    ``Agent(tools=[codex_mcp(allow=["*"])])``.

    !!! warning
        Codex's MCP-server interface is **experimental** (per OpenAI's docs)
        and may change without notice. Pin your Codex version if you depend
        on the exposed tool shape.

    ``allow=`` (or ``deny=``) is **required** — deny-by-default. Patterns match
    the namespaced tool name (``codex.*``). Tool names are not hardcoded because
    they depend on the installed Codex version; discover them with
    ``allow=["*"]`` once.

    Parameters
    ----------
    name:
        Server name and default namespace prefix (``"codex"``).
    allow / deny:
        fnmatch globs against the namespaced tool name (deny-by-default).
    args:
        Extra args appended after ``mcp-server`` (rarely needed).
    env:
        Extra environment for the subprocess. Auth is otherwise inherited
        from ``codex login`` / the current shell environment.
    namespace / prefix / cache_tools_ttl:
        Forwarded to :meth:`MCP.stdio` unchanged.

    Requires the ``mcp`` extra: ``pip install "lazytoolkit[mcp] @ git+https://github.com/selvaz/LazyTools.git"``.
    """
    # resolve_codex_bin() also finds the Codex desktop app's un-PATH'd install
    # dir, but it resolves against *this* process's environment. A caller
    # passing env={"PATH": ...} to MCP.stdio is explicitly selecting which
    # Codex install the child subprocess should see, so that override must be
    # searched explicitly rather than resolved from the parent's environment
    # -- and, on Windows, rather than left for the child to resolve on its
    # own: CreateProcess resolves a bare command name against the *calling*
    # process's PATH/search rules, not the child env's PATH, so a bare
    # "codex" would silently launch whatever this process's real PATH finds
    # instead of failing, ignoring the override either way. Verified live:
    # subprocess.run(["foo"], env={"PATH": <dir containing only foo.cmd>})
    # raises FileNotFoundError on this platform. The same reasoning applies
    # if the override itself finds nothing: falling back to a bare "codex"
    # would let CreateProcess silently resolve a *different* install from
    # this process's real PATH, so that failure is raised instead.
    #
    # PATH's key is matched case-insensitively only on Windows, where env var
    # names are themselves case-insensitive and "Path" is the common
    # spelling: on POSIX, names are case-sensitive, so an unrelated "Path"
    # entry must not be misread as a PATH override.
    key_matches = (lambda k: k.upper() == "PATH") if os.name == "nt" else (lambda k: k == "PATH")
    path_override = next((v for k, v in (env or {}).items() if key_matches(k)), None)
    if path_override is not None:
        resolved = shutil.which("codex", path=path_override)
        if resolved is None:
            raise FileNotFoundError(f"codex CLI not found on the overridden PATH {path_override!r}")
        # shutil.which() can return a relative path when path_override itself
        # has a relative entry (e.g. "tools" or "."). MCP.stdio launches the
        # subprocess lazily, so a relative command's meaning could change if
        # the process's cwd changes between now and the first tool call --
        # normalize against *this* moment's cwd instead.
        command = os.path.abspath(resolved)
    else:
        command = resolve_codex_bin() or "codex"
    return MCP.stdio(
        name,
        command=command,
        args=["mcp-server", *(args or [])],
        env=env,
        allow=allow,
        deny=deny,
        namespace=namespace,
        prefix=prefix,
        cache_tools_ttl=cache_tools_ttl,
    )

lazytools.connectors.code_support.codex_reviewer ¶

codex_reviewer(*, root: str | None = None, model: str | None = None, effort: str | None = None, timeout: float = DEFAULT_REVIEW_TIMEOUT, name: str = 'codex_code_review', system: str = CODE_REVIEWER_SYSTEM) -> Tool

Build the codex_code_review tool: one Codex-engined review agent.

root confines every call's repo_path (default: LAZYTOOLS_CODE_ROOT or the current working directory). model/effort default to whatever the local Codex CLI is configured with (~/.codex/config.toml) when left None. timeout bounds one review; it is passed as both the App Server request timeout and (at two thirds) the stream-idle timeout.

Raises ValueError on a non-positive / non-finite timeout — the MCP provider validates its env var, and the direct API must not be the lax way in: a negative value builds a tool whose every call dies inside CodexEngine, and inf removes the ceiling this parameter advertises.

Raises FileNotFoundError if the codex CLI cannot be located, so a caller that builds providers defensively (the MCP server's default_providers) skips the tool instead of serving one that fails on every call.

Source code in src/lazytools/connectors/code_support/_review.py

def codex_reviewer(
    *,
    root: str | None = None,
    model: str | None = None,
    effort: str | None = None,
    timeout: float = DEFAULT_REVIEW_TIMEOUT,
    name: str = "codex_code_review",
    system: str = CODE_REVIEWER_SYSTEM,
) -> Tool:
    """Build the ``codex_code_review`` tool: one Codex-engined review agent.

    ``root`` confines every call's ``repo_path`` (default: ``LAZYTOOLS_CODE_ROOT``
    or the current working directory). ``model``/``effort`` default to whatever
    the local Codex CLI is configured with (``~/.codex/config.toml``) when left
    ``None``. ``timeout`` bounds one review; it is passed as both the App Server
    request timeout and (at two thirds) the stream-idle timeout.

    Raises ``ValueError`` on a non-positive / non-finite ``timeout`` — the MCP
    provider validates its env var, and the direct API must not be the lax way
    in: a negative value builds a tool whose every call dies inside
    ``CodexEngine``, and ``inf`` removes the ceiling this parameter advertises.

    Raises ``FileNotFoundError`` if the ``codex`` CLI cannot be located, so a
    caller that builds providers defensively (the MCP server's
    ``default_providers``) skips the tool instead of serving one that fails on
    every call.
    """
    from lazybridge import Tool
    from lazybridge.engines.codex import codex_executable

    if not math.isfinite(timeout) or timeout <= 0:
        raise ValueError(f"timeout must be a positive finite number of seconds, got {timeout!r}")
    codex_executable()  # fail fast: no CLI, no tool

    # Resolved once, here: a relative root kept as-is would be re-resolved
    # against the process cwd on every call, so a later chdir would silently
    # move the boundary this tool is confined to.
    base = Path(root or os.environ.get("LAZYTOOLS_CODE_ROOT") or Path.cwd()).expanduser().resolve()

    async def codex_code_review(
        task: str,
        repo_path: str | None = None,
        diff_ref: str | None = None,
        paths: str | None = None,
        thread_id: str | None = None,
    ) -> str:
        """Have Codex review code in a local repository and report defects.

        Runs a real code review in a read-only sandbox: Codex reads the files
        itself and can run `git diff` / `git log`, so point it at a repository
        and say what to look at. Returns the review as text; it never modifies
        anything. Slow (tens of seconds to several minutes) and it costs a
        Codex turn, so ask one focused question per call.

        The header of every reply carries `thread_id=<id>`. Pass it back to ask
        a follow-up in the SAME Codex conversation: it still knows what it read
        and concluded, so the follow-up skips re-exploring the repository.
        Omit it to start fresh.

        Args:
            task: What to review and what to look for, e.g. "review the error
                handling in src/foo/bar.py" or "is the new retry logic
                correct?". Include any context the reviewer cannot infer.
            repo_path: Repository (or subdirectory) to review, absolute or
                relative to the server's code root. Defaults to the root.
            diff_ref: Git ref to review changes against, e.g. "main" or
                "HEAD~1". When set, the review is scoped to that diff plus
                uncommitted work.
            paths: Optional comma-separated paths to restrict the review to.
                Each must live inside the reviewed repository.
            thread_id: Continue an earlier review conversation, from the
                `thread_id=` in its reply. A thread belongs to the repository
                it was opened on — don't reuse one against a different repo.
        """
        cwd = _resolve_repo(repo_path, base)
        scoped = _confine_paths(paths, cwd)
        scope = _scope_block(diff_ref, scoped)
        return await _turn(
            label=name,
            prompt=f"{task}\n\n{scope}" if scope else task,
            cwd=cwd,
            base=base,
            system=system,
            agent_name="codex-code-reviewer",
            thread_id=thread_id,
            model=model,
            effort=effort,
            timeout=timeout,
        )

    return Tool(codex_code_review, name=name)

lazytools.connectors.code_support.codex_consultant ¶

codex_consultant(*, root: str | None = None, model: str | None = None, effort: str | None = None, timeout: float = DEFAULT_REVIEW_TIMEOUT, name: str = 'codex_ask', system: str = CODE_CONSULTANT_SYSTEM, tools: list[Any] | None = None) -> Tool

Build the codex_ask tool: Codex as a design partner, not a reviewer.

Same engine, same confinement and durable-thread handling as :func:codex_reviewer; what differs is the role. The reviewer is pointed at code and asked what is wrong with it; this one is asked a question — "does this protocol support X", "what breaks if I do Y" — and is told to separate what it verified in the source from what it is inferring, and to answer "I don't know, here is the experiment that would settle it" when that is the truth.

It exists because the reviewer prompt is the wrong instrument for a design conversation: asked a question, it answers with a findings list.

model/effort here are the defaults; each codex_ask call may override them, because a consultation is exactly the place where "same question, stronger model" is a legitimate move. tools are extra LazyBridge tools handed to the agent as Codex dynamic tools (e.g. the LazyCrawler web tools) — a consultant may need to read the world, where a reviewer only needs to read the repository.

Source code in src/lazytools/connectors/code_support/_review.py

def codex_consultant(
    *,
    root: str | None = None,
    model: str | None = None,
    effort: str | None = None,
    timeout: float = DEFAULT_REVIEW_TIMEOUT,
    name: str = "codex_ask",
    system: str = CODE_CONSULTANT_SYSTEM,
    tools: list[Any] | None = None,
) -> Tool:
    """Build the ``codex_ask`` tool: Codex as a design partner, not a reviewer.

    Same engine, same confinement and durable-thread handling as
    :func:`codex_reviewer`; what differs is the role. The reviewer is pointed at
    code and asked what is wrong with it; this one is asked a question — "does
    this protocol support X", "what breaks if I do Y" — and is told to separate
    what it verified in the source from what it is inferring, and to answer "I
    don't know, here is the experiment that would settle it" when that is the
    truth.

    It exists because the reviewer prompt is the wrong instrument for a design
    conversation: asked a question, it answers with a findings list.

    ``model``/``effort`` here are the *defaults*; each ``codex_ask`` call may
    override them, because a consultation is exactly the place where "same
    question, stronger model" is a legitimate move. ``tools`` are extra
    LazyBridge tools handed to the agent as Codex dynamic tools (e.g. the
    LazyCrawler web tools) — a consultant may need to read the world, where a
    reviewer only needs to read the repository.
    """
    from lazybridge import Tool
    from lazybridge.engines.codex import codex_executable

    if not math.isfinite(timeout) or timeout <= 0:
        raise ValueError(f"timeout must be a positive finite number of seconds, got {timeout!r}")
    codex_executable()

    base = Path(root or os.environ.get("LAZYTOOLS_CODE_ROOT") or Path.cwd()).expanduser().resolve()
    default_model, default_effort = model, effort
    extra_tools = list(tools or [])

    async def codex_ask(
        question: str,
        repo_path: str | None = None,
        thread_id: str | None = None,
        model: str | None = None,
        effort: str | None = None,
    ) -> str:
        """Ask Codex a technical question about a local repository.

        A second opinion from a different model that reads the code itself:
        design trade-offs, "is this protocol/API able to do X", "what would
        break if I changed Y". Read-only on the repository — it answers, it
        never edits — but it can also search and read the web when its tools
        include them. Slow (tens of seconds to several minutes) and it costs a
        Codex turn.

        It has none of your conversation context, so state the question
        self-containedly: what you are building, what you already know, and
        what you actually want decided.

        The header of every reply carries `thread_id=<id>`. Pass it back to
        continue the SAME conversation — Codex keeps what it read and
        concluded, so the follow-up is much cheaper than restating everything.

        Args:
            question: The question, with enough context to answer it.
            repo_path: Repository the question is about, absolute or relative
                to the server's code root. Defaults to the root.
            thread_id: Continue an earlier conversation, from the `thread_id=`
                in its reply. A thread belongs to the repository it was opened
                on — don't reuse one against a different repo.
            model: Codex model override for this call (e.g. "gpt-5.6-sol").
                Defaults to the server's configured model.
            effort: Reasoning effort override for this call ("low", "medium",
                "high", "xhigh"). Defaults to the server's configured effort.
        """
        cwd = _resolve_repo(repo_path, base)
        return await _turn(
            label=name,
            prompt=question,
            cwd=cwd,
            base=base,
            system=system,
            agent_name="codex-design-partner",
            thread_id=thread_id,
            model=model or default_model,
            effort=effort or default_effort,
            timeout=timeout,
            tools=extra_tools,
        )

    return Tool(codex_ask, name=name)

lazytools.connectors.code_support.codex_native_reviewer ¶

codex_native_reviewer(*, root: str | None = None, model: str | None = None, effort: str | None = None, timeout: float = DEFAULT_REVIEW_TIMEOUT, name: str = 'codex_review_changes') -> Tool

Build codex_review_changes: Codex' OWN review harness, typed target.

Where :func:codex_reviewer sends a prompt (steerable, and the only option for "look at this specific question"), this one calls review/start with a typed target and gets the harness Codex ships for reviewing diffs — severity-tagged findings with file:line, and no prompt of ours in the way.

The trade is exactly that: the protocol has no prompt slot, so the review cannot be steered. Use it for "review this branch, your standards"; use codex_code_review when you have a question.

The review runs inline on a durable thread, so the returned thread_id can be handed to codex_ask to interrogate the findings afterwards.

Source code in src/lazytools/connectors/code_support/_review.py

def codex_native_reviewer(
    *,
    root: str | None = None,
    model: str | None = None,
    effort: str | None = None,
    timeout: float = DEFAULT_REVIEW_TIMEOUT,
    name: str = "codex_review_changes",
) -> Tool:
    """Build ``codex_review_changes``: Codex' OWN review harness, typed target.

    Where :func:`codex_reviewer` sends a prompt (steerable, and the only option
    for "look at this specific question"), this one calls ``review/start`` with
    a typed target and gets the harness Codex ships for reviewing diffs —
    severity-tagged findings with file:line, and no prompt of ours in the way.

    The trade is exactly that: the protocol has no prompt slot, so the review
    **cannot be steered**. Use it for "review this branch, your standards"; use
    ``codex_code_review`` when you have a question.

    The review runs inline on a durable thread, so the returned ``thread_id``
    can be handed to ``codex_ask`` to interrogate the findings afterwards.
    """
    from lazybridge import Tool
    from lazybridge.engines.codex import codex_executable

    if not math.isfinite(timeout) or timeout <= 0:
        raise ValueError(f"timeout must be a positive finite number of seconds, got {timeout!r}")
    codex_executable()

    base = Path(root or os.environ.get("LAZYTOOLS_CODE_ROOT") or Path.cwd()).expanduser().resolve()

    async def codex_review_changes(
        repo_path: str | None = None,
        scope: str = "uncommitted",
        ref: str | None = None,
    ) -> str:
        """Review a diff with Codex' built-in review harness.

        Unlike `codex_code_review`, this takes no instructions: Codex reviews
        the changes by its own standards and returns findings tagged by
        severity ([P1] worst) with file:line. Use it for an unsteered second
        opinion on a branch or commit; use `codex_code_review` when you need to
        ask about something specific. Slow (minutes) and costs a Codex turn.

        The reply header carries `thread_id=<handle>`; pass it to `codex_ask`
        to question the findings without the review being run again.

        Args:
            repo_path: Repository to review, absolute or relative to the
                server's code root. Defaults to the root.
            scope: What to review — "uncommitted" (staged, unstaged and
                untracked work), "branch" (this branch against `ref`), or
                "commit" (the single commit `ref`).
            ref: The base branch for scope="branch" (e.g. "main"), or the sha
                for scope="commit". Ignored when scope="uncommitted".
        """
        if scope not in _REVIEW_TARGETS:
            raise ValueError(f"scope must be one of {', '.join(_REVIEW_TARGETS)}, got {scope!r}")
        if scope != "uncommitted" and not ref:
            raise ValueError(f"scope={scope!r} needs a ref (a base branch, or a commit sha)")
        cwd = _resolve_repo(repo_path, base)
        return await _turn(
            label=name,
            prompt="",  # not sent: review/start has no prompt slot
            cwd=cwd,
            base=base,
            system=CODE_REVIEWER_SYSTEM,
            agent_name="codex-native-reviewer",
            thread_id=None,
            model=model,
            effort=effort,
            timeout=timeout,
            review_target=_REVIEW_TARGETS[scope](ref),
        )

    return Tool(codex_review_changes, name=name)

lazytools.connectors.code_support.build_cli_collaboration ¶

build_cli_collaboration(*, name: str = 'cli_collaboration', description: str | None = None, claude_model: str = 'claude-opus-4-8', codex_model: str = 'gpt-5.4', synthesizer_model: str = 'claude-opus-4-8', executor_model: str = 'claude-opus-4-8', execute: bool = False, base_dir: str | None = None, writer: CodeWriteTools | None = None) -> Agent

Build the Claude Code + Codex collaboration pipeline as a reusable tool.

Returns a named :class:~lazybridge.Agent (Plan engine) that you drop straight into Agent(tools=[build_cli_collaboration()]) — the same way you pass the :func:claude_code / :func:codex function tools. Because an Agent is a tool in LazyBridge, the whole multi-agent pipeline appears to the parent agent as a single callable taking one task string.

The default is read-only: three sessions — two read-only CLI analysts and one synthesizer that writes the plan. The codebase is never modified unless you opt in with execute=True + base_dir=.

Parameters¶

name: Tool name the parent agent sees. Must be explicit (used as the tool-map key); defaults to "cli_collaboration". description: Tool description shown to the parent LLM. Defaults to a summary of the pipeline's behaviour. claude_model: Model driving the Claude-Code analyst (step 1, read-only). codex_model: Model driving the Codex analyst/critic (step 2, read-only). synthesizer_model: Model that merges the two analyses into one plan (step 3). executor_model: Model that implements the plan via the gated writer (step 4, only when execute=True). execute: Default False — the pipeline ends at the written plan. Pass True (with base_dir=) to append the executor step, which implements the plan via claude_code_write sandboxed to base_dir. base_dir: Convenience for execute=True: builds an internal :class:CodeWriteTools sandboxed to this root with require_confirmation=False — the pipeline is autonomous, so the sandbox (plus a git checkout) is the safety rail. Mutually exclusive with writer=. writer: Bring your own :class:CodeWriteTools for the executor instead. This is the only way to run the executor with confirmation gating: you hold the instance, so you can call writer.confirm_write() (once per executor write call) while the pipeline runs — a gate-enabled writer built inside this function would block forever, since nobody could reach it to grant.

Notes¶

The claude_analyst writes its analysis into a shared Memory that the codex_analyst reads via sources=. This is safe because Plan runs steps strictly sequentially — there is no concurrent writer/reader race.

Source code in src/lazytools/connectors/code_support/_collaboration.py

def build_cli_collaboration(
    *,
    name: str = "cli_collaboration",
    description: str | None = None,
    claude_model: str = "claude-opus-4-8",
    codex_model: str = "gpt-5.4",
    synthesizer_model: str = "claude-opus-4-8",
    executor_model: str = "claude-opus-4-8",
    execute: bool = False,
    base_dir: str | None = None,
    writer: CodeWriteTools | None = None,
) -> Agent:
    """Build the Claude Code + Codex collaboration pipeline as a reusable tool.

    Returns a named :class:`~lazybridge.Agent` (``Plan`` engine) that you drop
    straight into ``Agent(tools=[build_cli_collaboration()])`` — the same way you
    pass the :func:`claude_code` / :func:`codex` function tools. Because an
    ``Agent`` *is* a tool in LazyBridge, the whole multi-agent pipeline appears
    to the parent agent as a single callable taking one ``task`` string.

    **The default is read-only**: three sessions — two read-only CLI analysts
    and one synthesizer that writes the *plan*. The codebase is never modified
    unless you opt in with ``execute=True`` + ``base_dir=``.

    Parameters
    ----------
    name:
        Tool name the parent agent sees. Must be explicit (used as the tool-map
        key); defaults to ``"cli_collaboration"``.
    description:
        Tool description shown to the parent LLM. Defaults to a summary of the
        pipeline's behaviour.
    claude_model:
        Model driving the Claude-Code analyst (step 1, read-only).
    codex_model:
        Model driving the Codex analyst/critic (step 2, read-only).
    synthesizer_model:
        Model that merges the two analyses into one plan (step 3).
    executor_model:
        Model that implements the plan via the gated writer (step 4, only
        when ``execute=True``).
    execute:
        Default ``False`` — the pipeline ends at the written plan. Pass
        ``True`` (with ``base_dir=``) to append the executor step, which
        implements the plan via ``claude_code_write`` sandboxed to
        ``base_dir``.
    base_dir:
        Convenience for ``execute=True``: builds an internal
        :class:`CodeWriteTools` sandboxed to this root with
        ``require_confirmation=False`` — the pipeline is autonomous, so the
        sandbox (plus a git checkout) is the safety rail. Mutually exclusive
        with ``writer=``.
    writer:
        Bring your own :class:`CodeWriteTools` for the executor instead.
        This is the only way to run the executor with confirmation gating:
        you hold the instance, so you can call ``writer.confirm_write()``
        (once per executor write call) while the pipeline runs — a
        gate-enabled writer built *inside* this function would block
        forever, since nobody could reach it to grant.

    Notes
    -----
    The ``claude_analyst`` writes its analysis into a shared ``Memory`` that the
    ``codex_analyst`` reads via ``sources=``. This is safe because ``Plan`` runs
    steps strictly sequentially — there is no concurrent writer/reader race.
    """
    # Deferred imports: keep module import stdlib-light (see module docstring).
    from lazybridge import Agent, LLMEngine, Memory, Plan, Step, from_step

    # DeduplicateGuard shipped after lazybridge 0.9.0; degrade gracefully on
    # older installs (the guard is an optimisation — it stops an analyst from
    # re-issuing an identical CLI call — not a correctness requirement).
    try:
        from lazybridge import DeduplicateGuard  # type: ignore[attr-defined]

        def _dedup():
            return DeduplicateGuard(verbose=False)
    except ImportError:  # lazybridge <= 0.9.0

        def _dedup():
            return None

    if execute:
        if writer is not None and base_dir is not None:
            raise ValueError(
                "build_cli_collaboration: pass either writer= (your own CodeWriteTools, "
                "you keep the confirm_write() handle) or base_dir= (internal ungated "
                "writer), not both."
            )
        if writer is None and base_dir is None:
            raise ValueError(
                "build_cli_collaboration(execute=True) requires base_dir= (internal "
                "ungated writer sandboxed there) or writer= (your own CodeWriteTools). "
                "Omit execute (default False) for the read-only analyse+plan pipeline."
            )
    elif writer is not None or base_dir is not None:
        raise ValueError(
            "build_cli_collaboration: writer=/base_dir= only apply with execute=True "
            "(the default pipeline is read-only and never writes)."
        )

    # Shared dialogue: claude_analyst writes (memory=), codex_analyst reads
    # (sources=). Safe under Plan's sequential execution — no parallel access.
    dialogue = Memory(strategy="summary")

    claude_analyst = Agent(
        name="claude_analyst",
        engine=LLMEngine(
            claude_model,
            tool_timeout=None,
            system=(
                "Analyse the task using claude_code in mode='read'. "
                "Propose a concrete implementation approach. Be concise."
            ),
        ),
        tools=[claude_code],
        memory=dialogue,
        guard=_dedup(),
    )

    codex_analyst = Agent(
        name="codex_analyst",
        engine=LLMEngine(
            codex_model,
            tool_timeout=None,
            system=(
                "Analyse the task using the read-only codex tool. "
                "Critique or confirm claude_analyst's approach. Be concise."
            ),
        ),
        tools=[codex],
        sources=[dialogue],  # sees claude_analyst's analysis as context
        guard=_dedup(),
    )

    synthesizer = Agent(
        name="synthesizer",
        engine=LLMEngine(
            synthesizer_model,
            system=(
                "You receive two code analyses (Claude Code and Codex). "
                "Produce a single, concrete, step-by-step implementation plan."
            ),
        ),
    )

    steps = [
        Step("claude_analyst"),
        Step("codex_analyst", context=from_step("claude_analyst")),
        Step("synthesizer", context=from_step("codex_analyst")),
    ]
    # Annotated as the Agent(tools=) element type: a bare list[Agent] is
    # rejected because list is invariant against that wider union.
    tools: list[Any] = [claude_analyst, codex_analyst, synthesizer]

    if execute:
        if writer is None:
            assert base_dir is not None  # narrowed by the ValueError above
            writer = CodeWriteTools(
                base_dir=base_dir,
                claude=True,
                codex=False,
                require_confirmation=False,
            )
        executor = Agent(
            name="executor",
            engine=LLMEngine(
                executor_model,
                tool_timeout=None,
                system="Implement the plan you receive using the claude_code_write tool.",
            ),
            tools=[*writer.as_tools()],
        )
        steps.append(Step("executor", context=from_step("synthesizer")))
        tools.append(executor)

    return Agent(
        name=name,
        description=description or _DEFAULT_DESCRIPTION,
        engine=Plan(*steps),
        tools=tools,
    )

lazytools.connectors.code_support.check_clis_available ¶

check_clis_available() -> dict[str, bool]

Return availability of the 'claude' and 'codex' CLIs.

Returns a {"claude": bool, "codex": bool} dict. Call this at startup to surface missing CLIs immediately rather than at the first tool call.

codex is resolved via :func:resolve_codex_bin — not a bare shutil.which("codex") — so a Codex install reachable only through its desktop app's un-PATH'd directory is correctly reported as available, matching what codex()/codex_write will actually be able to run.

Source code in src/lazytools/connectors/code_support/__init__.py

def check_clis_available() -> dict[str, bool]:
    """Return availability of the 'claude' and 'codex' CLIs.

    Returns a ``{"claude": bool, "codex": bool}`` dict. Call this at startup
    to surface missing CLIs immediately rather than at the first tool call.

    ``codex`` is resolved via :func:`resolve_codex_bin` — not a bare
    ``shutil.which("codex")`` — so a Codex install reachable only through its
    desktop app's un-``PATH``'d directory is correctly reported as available,
    matching what ``codex()``/``codex_write`` will actually be able to run.
    """
    return {
        "claude": shutil.which("claude") is not None,
        "codex": resolve_codex_bin() is not None,
    }

Gateway¶

lazytools.connectors.gateway.ExternalToolProvider ¶

ExternalToolProvider(client: ExternalToolClient, *, specs: Iterable[ExternalToolSpec | Mapping[str, Any]] | None = None, include: Iterable[str] | None = None, exclude: Iterable[str] | None = None, name_prefix: str = '', strict: bool | None = None)

Expose an external tool registry as a LazyBridge tool provider.

Agent(tools=[ExternalToolProvider(client)]) expands the provider into normal :class:Tool objects through LazyBridge's existing _is_lazy_tool_provider hook.

Source code in src/lazytools/connectors/gateway/__init__.py

def __init__(
    self,
    client: ExternalToolClient,
    *,
    specs: Iterable[ExternalToolSpec | Mapping[str, Any]] | None = None,
    include: Iterable[str] | None = None,
    exclude: Iterable[str] | None = None,
    name_prefix: str = "",
    strict: bool | None = None,
) -> None:
    self.client = client
    self._specs = list(specs) if specs is not None else None
    self.include = set(include or []) or None
    self.exclude = set(exclude or [])
    self.name_prefix = name_prefix
    self.strict = strict

lazytools.connectors.gateway.JsonHttpExternalToolClient ¶

JsonHttpExternalToolClient(base_url: str, *, api_key: str | None = None, headers: Mapping[str, str] | None = None, timeout: float = 30.0, tools_path: str = '/tools', call_path_template: str = '/tools/{name}/call')

Small stdlib HTTP client for a JSON external-tool gateway.

Default endpoint contract: - GET {base_url}/tools returns either [{...}] or {"tools": [{...}]}. - POST {base_url}/tools/{name}/call with {"arguments": {...}} returns JSON.

This class is intentionally narrow. For Pipedream/Composio/Arcade, wrap their SDK/API behind :class:ExternalToolClient when their HTTP shape differs from the default contract.

Source code in src/lazytools/connectors/gateway/__init__.py

def __init__(
    self,
    base_url: str,
    *,
    api_key: str | None = None,
    headers: Mapping[str, str] | None = None,
    timeout: float = 30.0,
    tools_path: str = "/tools",
    call_path_template: str = "/tools/{name}/call",
) -> None:
    self.base_url = base_url.rstrip("/")
    self.api_key = api_key
    self.headers = dict(headers or {})
    self.timeout = timeout
    self.tools_path = tools_path
    self.call_path_template = call_path_template
    # Never send a bearer credential in cleartext. Plain HTTP is allowed
    # only for loopback (local development gateways).
    parsed = urllib.parse.urlsplit(self.base_url)
    if api_key and parsed.scheme != "https" and parsed.hostname not in {"localhost", "127.0.0.1", "::1"}:
        raise ValueError(
            f"JsonHttpExternalToolClient: refusing to send api_key over "
            f"{parsed.scheme or '<no scheme>'} to {parsed.hostname!r} — use an https:// "
            f"base_url (plain http is permitted only for localhost)."
        )

lazytools.connectors.gateway.ExternalToolSpec `dataclass` ¶

ExternalToolSpec(name: str, description: str, parameters: Mapping[str, Any], strict: bool = False)

A remotely hosted tool definition.

parameters is the provider-agnostic JSON Schema object used by LazyBridge providers when advertising tools to an LLM.

from_mapping `classmethod` ¶

from_mapping(raw: Mapping[str, Any]) -> ExternalToolSpec

Build a spec from common external registry shapes.

Accepted inputs: - {"name", "description", "parameters"} - OpenAI-style {"function": {"name", "description", "parameters"}}

Source code in src/lazytools/connectors/gateway/__init__.py

@classmethod
def from_mapping(cls, raw: Mapping[str, Any]) -> ExternalToolSpec:
    """Build a spec from common external registry shapes.

    Accepted inputs:
    - ``{"name", "description", "parameters"}``
    - OpenAI-style ``{"function": {"name", "description", "parameters"}}``
    """
    data: Mapping[str, Any]
    if isinstance(raw.get("function"), Mapping):
        data = raw["function"]  # type: ignore[index]
    else:
        data = raw

    name = data.get("name")
    if not isinstance(name, str) or not name:
        raise ValueError("External tool spec must include a non-empty string name")

    description = data.get("description")
    if description is None:
        description = f"Call external tool {name}."
    if not isinstance(description, str):
        raise ValueError(f"External tool {name!r} description must be a string")

    parameters = data.get("parameters") or _JSON_OBJECT_SCHEMA
    if not isinstance(parameters, Mapping):
        raise ValueError(f"External tool {name!r} parameters must be a JSON Schema object")

    strict = bool(data.get("strict", raw.get("strict", False)))
    return cls(name=name, description=description, parameters=parameters, strict=strict)

SEC EDGAR (transport client only)¶

Agents reach SEC and market data through the hub-backed datahub_* tools (see Financial data). EdgarClient is a low-level transport client for non-agent code only — it is not an agent tool provider.

lazytools.connectors.edgar.EdgarClient ¶

EdgarClient(user_agent: str, *, http: Any | None = None, timeout: float = 30.0, max_response_bytes: int = DEFAULT_MAX_RESPONSE_BYTES, min_request_interval: float = DEFAULT_MIN_REQUEST_INTERVAL)

Production :class:EdgarService backed by the SEC EDGAR APIs over HTTPS.

Parameters:

Name	Type	Description	Default
`user_agent`	`str`	Required. A declared identity per the SEC fair-access policy, e.g. `"Jane Doe jane@example.com"`. Empty/blank raises `ValueError`.	required
`http`	`Any \| None`	Optional injected HTTP client (an `httpx.Client` or anything exposing `stream(method, url, headers=...)`). When omitted, an `httpx.Client` is built lazily on first use.	`None`
`timeout`	`float`	Request timeout in seconds for the lazily built client.	`30.0`
`max_response_bytes`	`int`	Hard cap applied to every response body.	`DEFAULT_MAX_RESPONSE_BYTES`
`min_request_interval`	`float`	Minimum spacing between requests in seconds (simple monotonic-clock throttle); `0` disables the throttle.	`DEFAULT_MIN_REQUEST_INTERVAL`

Source code in src/lazytools/connectors/edgar/client.py

def __init__(
    self,
    user_agent: str,
    *,
    http: Any | None = None,
    timeout: float = 30.0,
    max_response_bytes: int = DEFAULT_MAX_RESPONSE_BYTES,
    min_request_interval: float = DEFAULT_MIN_REQUEST_INTERVAL,
) -> None:
    if not user_agent or not user_agent.strip():
        raise ValueError(
            "EdgarClient requires a non-empty user_agent. The SEC fair-access policy "
            "requires a declared User-Agent identifying you, e.g. 'Jane Doe jane@example.com'."
        )
    self._user_agent = user_agent.strip()
    self._http = http
    self._timeout = timeout
    self._max_response_bytes = max_response_bytes
    self._min_request_interval = min_request_interval
    self._last_request_at: float | None = None
    # company_tickers.json cached in-memory for the client's lifetime.
    self._tickers_cache: dict[str, Any] | None = None

resolve_company ¶

resolve_company(query: str, *, limit: int = 10) -> list[dict[str, str]]

Resolve a ticker or company-name query against company_tickers.json.

Exact (case-insensitive) ticker matches come first, then substring matches on the company title. Each entry is {"cik": "0000320193", "ticker": "AAPL", "title": "Apple Inc."} with the CIK zero-padded to 10 digits.

Source code in src/lazytools/connectors/edgar/client.py

def resolve_company(self, query: str, *, limit: int = 10) -> list[dict[str, str]]:
    """Resolve a ticker or company-name query against company_tickers.json.

    Exact (case-insensitive) ticker matches come first, then substring
    matches on the company title. Each entry is
    ``{"cik": "0000320193", "ticker": "AAPL", "title": "Apple Inc."}``
    with the CIK zero-padded to 10 digits.
    """
    q = query.strip().lower()
    if not q:
        raise ValueError("resolve_company requires a non-empty query")
    if self._tickers_cache is None:
        self._tickers_cache = self._get_json(_COMPANY_TICKERS_URL)
    exact: list[dict[str, str]] = []
    partial: list[dict[str, str]] = []
    for entry in self._tickers_cache.values():
        ticker = str(entry.get("ticker", ""))
        title = str(entry.get("title", ""))
        record = {"cik": str(entry.get("cik_str", "")).zfill(10), "ticker": ticker, "title": title}
        if ticker.lower() == q:
            exact.append(record)
        elif q in title.lower():
            partial.append(record)
    return (exact + partial)[:limit]

list_filings ¶

list_filings(cik: str, *, form: str | None = None, limit: int = 20) -> list[dict[str, Any]]

List a company's recent filings (newest first), optionally by form.

Reads the filings.recent arrays of the submissions JSON. Each entry carries accession_no, form, filed_at, report_date (None when EDGAR reports an empty string -- and note this is the covered PERIOD, e.g. a quarter-end, not the submission date; it is not a substitute for filed_at when checking filing recency), items (an 8-K's own item codes, e.g. ["2.02", "9.01"] for a results-of-operations 8-K; empty for every other form), primary_document, and the Archives url of the primary document.

Source code in src/lazytools/connectors/edgar/client.py

def list_filings(self, cik: str, *, form: str | None = None, limit: int = 20) -> list[dict[str, Any]]:
    """List a company's recent filings (newest first), optionally by form.

    Reads the ``filings.recent`` arrays of the submissions JSON. Each
    entry carries ``accession_no``, ``form``, ``filed_at``, ``report_date``
    (``None`` when EDGAR reports an empty string -- and note this is the
    covered PERIOD, e.g. a quarter-end, not the submission date; it is
    not a substitute for ``filed_at`` when checking filing recency),
    ``items`` (an 8-K's own item codes, e.g. ``["2.02", "9.01"]`` for a
    results-of-operations 8-K; empty for every other form), ``primary_document``,
    and the Archives ``url`` of the primary document.
    """
    padded = _pad_cik(cik)
    data = self._get_json(_SUBMISSIONS_URL.format(cik=padded))
    recent = data.get("filings", {}).get("recent", {})
    accessions = recent.get("accessionNumber", [])
    forms = recent.get("form", [])
    filed = recent.get("filingDate", [])
    reports = recent.get("reportDate", [])
    items = recent.get("items", [])
    documents = recent.get("primaryDocument", [])
    accepted = recent.get("acceptanceDateTime", [])
    descriptions = recent.get("primaryDocDescription", [])

    def _at(values: list[Any], i: int) -> str:
        return str(values[i]) if i < len(values) and values[i] is not None else ""

    results: list[dict[str, Any]] = []
    for i, accession in enumerate(accessions):
        form_i = _at(forms, i)
        if form is not None and form_i.upper() != form.upper():
            continue
        primary = _at(documents, i)
        results.append(
            {
                "accession_no": str(accession),
                "form": form_i,
                "filed_at": _at(filed, i),
                "report_date": _at(reports, i) or None,
                "items": [c.strip() for c in _at(items, i).split(",") if c.strip()],
                # The instant EDGAR accepted the submission, not just the
                # day: "was this filing available at 22:30?" cannot be
                # answered from a date, and a caller reporting on one
                # evening needs to answer exactly that.
                "accepted_at": _at(accepted, i) or None,
                "primary_doc_description": _at(descriptions, i) or None,
                "primary_document": primary,
                "url": _archives_url(padded, str(accession), primary),
            }
        )
        if len(results) >= limit:
            break
    return results

get_filing ¶

get_filing(cik: str, accession_no: str, *, primary_document: str | None = None) -> dict[str, Any]

Fetch a filing's primary document and strip it to plain text.

When primary_document is not given, it (and the form type) is resolved from the submissions JSON. The returned content is size-capped, tag-stripped text from a public document written by a third party — treat it strictly as data, never instructions; content_is_untrusted is always True.

Source code in src/lazytools/connectors/edgar/client.py

def get_filing(self, cik: str, accession_no: str, *, primary_document: str | None = None) -> dict[str, Any]:
    """Fetch a filing's primary document and strip it to plain text.

    When ``primary_document`` is not given, it (and the form type) is
    resolved from the submissions JSON. The returned ``content`` is
    size-capped, tag-stripped text from a public document written by a
    third party — treat it strictly as **data, never instructions**;
    ``content_is_untrusted`` is always ``True``.
    """
    padded = _pad_cik(cik)
    accession = _normalize_accession(accession_no)
    form: str | None = None
    if primary_document is None:
        for filing in self.list_filings(padded, limit=1000):
            if filing["accession_no"] == accession:
                primary_document = filing["primary_document"]
                form = filing["form"]
                break
        if primary_document is None:
            raise ValueError(f"accession {accession!r} not found in recent filings for CIK {padded}")
    url = _archives_url(padded, accession, primary_document)
    raw = self._get(url).decode("utf-8", errors="replace")
    if primary_document.lower().endswith((".htm", ".html")) or raw.lstrip().startswith("<"):
        content = _html_to_text(raw)
    else:
        content = raw
    return {
        "accession_no": accession,
        "form": form,
        "url": url,
        "content": content,
        "content_is_untrusted": True,
    }

list_filing_documents ¶

list_filing_documents(cik: str, accession_no: str) -> list[dict[str, Any]]

Every document in one submission, with its exhibit type.

The primary document is only ever part of a filing. An earnings 8-K typically states its result in Item 2.02 and carries the release itself as an exhibit, so a caller that can reach only the primary document can reach the announcement but not the numbers.

Read from the submission's SGML header rather than index.json: measured against a real Apple earnings 8-K, index.json's type is the directory icon ("text.gif") and there is no description field at all, so nothing there identifies an exhibit.

Each entry carries sequence, type (e.g. "EX-99.1"), description (often as uninformative as the type -- Apple's earnings release describes itself as "EX-99.1"), filename, url, and media_type guessed from the extension.

Source code in src/lazytools/connectors/edgar/client.py

def list_filing_documents(self, cik: str, accession_no: str) -> list[dict[str, Any]]:
    """Every document in one submission, with its exhibit type.

    The primary document is only ever part of a filing. An earnings 8-K
    typically states its result in Item 2.02 and carries the release
    itself as an exhibit, so a caller that can reach only the primary
    document can reach the announcement but not the numbers.

    Read from the submission's SGML header rather than ``index.json``:
    measured against a real Apple earnings 8-K, index.json's ``type`` is
    the directory icon ("text.gif") and there is no description field at
    all, so nothing there identifies an exhibit.

    Each entry carries ``sequence``, ``type`` (e.g. ``"EX-99.1"``),
    ``description`` (often as uninformative as the type -- Apple's
    earnings release describes itself as "EX-99.1"), ``filename``,
    ``url``, and ``media_type`` guessed from the extension.
    """
    padded = _pad_cik(cik)
    dashed = _normalize_accession(accession_no)
    url = _INDEX_HEADERS_URL.format(
        cik_int=int(padded), accession=dashed.replace("-", ""), dashed=dashed)
    # Unescaped first: the header is served inside an HTML page, so its
    # SGML tags arrive as &lt;TYPE&gt; and a naive parse finds nothing.
    raw = _html_unescape(self._get(url).decode("utf-8", errors="replace"))
    documents: list[dict[str, Any]] = []
    for block in re.findall(r"<DOCUMENT>(.*?)</DOCUMENT>", raw, re.S | re.I):
        filename = _sgml_field(block, "FILENAME")
        if not filename:
            continue
        documents.append({
            "sequence": _sgml_field(block, "SEQUENCE"),
            "type": _sgml_field(block, "TYPE"),
            "description": _sgml_field(block, "DESCRIPTION"),
            "filename": filename,
            "media_type": _media_type(filename),
            "url": _archives_url(padded, dashed, filename),
        })
    if not documents:
        # A real submission always contains at least its own primary
        # document, so an empty inventory is this parse failing, not the
        # filing being empty -- and returning [] would tell the caller
        # the second. Review found one shape where the header 404s
        # instead (a 1994 accession), which raises on its own; this
        # covers a 200 whose body we could not read.
        raise RuntimeError(
            f"no documents parsed from the submission header for {dashed}; "
            f"the filing index at {_archives_url(padded, dashed, '')} lists them"
        )
    return documents

get_filing_document ¶

get_filing_document(cik: str, accession_no: str, filename: str) -> dict[str, Any]

Fetch one named document from a submission, as text.

filename must be one this submission actually contains: it is checked against :meth:list_filing_documents rather than pasted into an Archives URL. A caller-supplied path would otherwise decide what this client fetches, which is not a decision a caller gets to make even against a host we pin.

Binary documents are not decoded into text -- an image or a PDF comes back with extraction_status saying so and empty content, rather than a page of replacement characters pretending to be prose.

Source code in src/lazytools/connectors/edgar/client.py

def get_filing_document(self, cik: str, accession_no: str, filename: str) -> dict[str, Any]:
    """Fetch one named document from a submission, as text.

    ``filename`` must be one this submission actually contains: it is
    checked against :meth:`list_filing_documents` rather than pasted into
    an Archives URL. A caller-supplied path would otherwise decide what
    this client fetches, which is not a decision a caller gets to make
    even against a host we pin.

    Binary documents are not decoded into text -- an image or a PDF comes
    back with ``extraction_status`` saying so and empty ``content``,
    rather than a page of replacement characters pretending to be prose.
    """
    padded = _pad_cik(cik)
    dashed = _normalize_accession(accession_no)
    inventory = {d["filename"]: d for d in self.list_filing_documents(padded, dashed)}
    entry = inventory.get(filename)
    if entry is None:
        raise ValueError(
            f"{filename!r} is not a document of filing {dashed}; "
            f"choose one of: {sorted(inventory)[:10]}"
        )
    media = entry["media_type"]
    if media not in _TEXT_MEDIA:
        # Not fetched at all. The inventory already says this is a JPEG or
        # a PDF, so downloading it to discard it spends a request against
        # the SEC's rate limit and the caller's deadline to learn what we
        # already knew. size_bytes is the inventory's, or None.
        return {
            "accession_no": dashed, "filename": filename,
            "type": entry["type"], "description": entry["description"],
            "url": entry["url"], "media_type": media,
            "content": "",
            "extraction_status": "unsupported",
            "size_bytes": entry.get("size_bytes"),
            "content_is_untrusted": True,
        }
    body = self._get(entry["url"])
    raw = body.decode("utf-8", errors="replace")
    content = _html_to_text(raw) if _looks_like_html(filename, raw) else raw
    return {
        "accession_no": dashed, "filename": filename,
        "type": entry["type"], "description": entry["description"],
        "url": entry["url"], "media_type": media,
        "content": content,
        "extraction_status": "ok",
        "size_bytes": len(body),
        "content_is_untrusted": True,
    }

company_facts ¶

company_facts(cik: str) -> dict[str, Any]

Return the raw XBRL companyfacts JSON for a company, untouched.

Source code in src/lazytools/connectors/edgar/client.py

def company_facts(self, cik: str) -> dict[str, Any]:
    """Return the raw XBRL companyfacts JSON for a company, untouched."""
    padded = _pad_cik(cik)
    return self._get_json(_COMPANY_FACTS_URL.format(cik=padded))

Market data (transport client only)¶

MarketDataClient is likewise transport-only plumbing, not an agent tool provider; agents use datahub_*.

lazytools.connectors.marketdata.MarketDataClient ¶

MarketDataClient(adapter: MarketDataAdapter)

Price lookups through a swappable :class:MarketDataAdapter.

Source code in src/lazytools/connectors/marketdata/client.py

def __init__(self, adapter: MarketDataAdapter) -> None:
    self._adapter = adapter

prices_get ¶

prices_get(ticker: str) -> dict[str, str]

Latest quote for a ticker.

Returns {"ticker": "AAPL", "price": "203.92", "currency": "USD", "as_of": "2026-06-09", "source": "stooq"} — the price is a string (Decimal-safe, see module docstring).

Source code in src/lazytools/connectors/marketdata/client.py

def prices_get(self, ticker: str) -> dict[str, str]:
    """Latest quote for a ticker.

    Returns ``{"ticker": "AAPL", "price": "203.92", "currency": "USD",
    "as_of": "2026-06-09", "source": "stooq"}`` — the price is a string
    (Decimal-safe, see module docstring).
    """
    symbol = ticker.strip()
    if not symbol:
        raise ValueError("ticker must be non-empty")
    result = self._adapter.quote(symbol)
    return {
        "ticker": symbol.upper(),
        "price": result["price"],
        "currency": result["currency"],
        "as_of": result["as_of"],
        "source": result["source"],
    }

prices_history ¶

prices_history(ticker: str, *, range_: str = '1y') -> list[dict[str, str]]

Daily OHLCV history for a ticker over range_.

range_ is one of "1m"/"3m"/"6m"/"1y"/"5y", filtered client-side by date. Each row is {"date", "open", "high", "low", "close", "volume"} with every value a string (Decimal-safe).

Source code in src/lazytools/connectors/marketdata/client.py

def prices_history(self, ticker: str, *, range_: str = "1y") -> list[dict[str, str]]:
    """Daily OHLCV history for a ticker over ``range_``.

    ``range_`` is one of ``"1m"``/``"3m"``/``"6m"``/``"1y"``/``"5y"``,
    filtered client-side by date. Each row is ``{"date", "open", "high",
    "low", "close", "volume"}`` with every value a string (Decimal-safe).
    """
    symbol = ticker.strip()
    if not symbol:
        raise ValueError("ticker must be non-empty")
    if range_ not in VALID_RANGES:
        raise ValueError(f"invalid range_ {range_!r}; expected one of {list(VALID_RANGES)}")
    return self._adapter.history(symbol, range_=range_)

lazytools.connectors.marketdata.StooqAdapter ¶

StooqAdapter(*, http: Any | None = None, timeout: float = 30.0, max_response_bytes: int = DEFAULT_MAX_RESPONSE_BYTES)

Free, key-less :class:MarketDataAdapter backed by stooq.com CSV endpoints.

US tickers map to stooq's {ticker}.us convention; a ticker that already carries a market suffix (sap.de) is passed through unchanged.

Parameters:

Name	Type	Description	Default
`http`	`Any \| None`	Optional injected HTTP client (an `httpx.Client` or anything exposing `stream(method, url)`). Built lazily when omitted.	`None`
`timeout`	`float`	Request timeout in seconds for the lazily built client.	`30.0`
`max_response_bytes`	`int`	Hard cap applied to every response body.	`DEFAULT_MAX_RESPONSE_BYTES`

Source code in src/lazytools/connectors/marketdata/adapters.py

def __init__(
    self,
    *,
    http: Any | None = None,
    timeout: float = 30.0,
    max_response_bytes: int = DEFAULT_MAX_RESPONSE_BYTES,
) -> None:
    self._http = http
    self._timeout = timeout
    self._max_response_bytes = max_response_bytes

Report (LazyReport)¶

lazytools.report.Memo ¶

Bases: BaseModel

A renderable memo/report: title, timestamp, sections, metadata.

lazytools.report.Section ¶

Bases: BaseModel

One memo section: a title, optional Markdown prose, optional tables and figures.

lazytools.report.TableBlock ¶

Bases: BaseModel

A simple rectangular table: header columns plus string-cell rows.

lazytools.report.FigureBlock ¶

Bases: BaseModel

A figure named by artifact ref; bytes are resolved only at render time.

ref is a canonical "scheme:key" string — the ecosystem's shared artifact identity (lazydatacore.ArtifactRef): regimes:<plot_key>, crawler:<content_hash>, chart:<spec>, file:<path>, bytes:<base64>. See :mod:lazytools.report.artifacts for resolution.

lazytools.report.render_markdown ¶

render_markdown(memo: Memo) -> str

Render a :class:Memo to GitHub-flavoured Markdown (deterministic).

Source code in src/lazytools/report/render.py

def render_markdown(memo: Memo) -> str:
    """Render a :class:`Memo` to GitHub-flavoured Markdown (deterministic)."""
    lines: list[str] = [f"# {memo.title}", ""]
    if memo.as_of is not None:
        lines += [f"_as of {memo.as_of.isoformat()}_", ""]
    for section in memo.sections:
        lines += [f"## {section.title}", ""]
        if section.body:
            lines += [section.body, ""]
        for table in section.tables:
            lines.append("| " + " | ".join(_md_cell(cell) for cell in table.columns) + " |")
            lines.append("| " + " | ".join("---" for _ in table.columns) + " |")
            for row in table.rows:
                lines.append("| " + " | ".join(_md_cell(cell) for cell in row) + " |")
            lines.append("")
        for figure in section.figures:
            caption = f"{figure.caption} " if figure.caption else ""
            lines += [f"_Figure: {caption}({figure.ref})_", ""]
    if memo.metadata:
        lines += ["---", ""]
        lines += [f"- {key}: {memo.metadata[key]}" for key in sorted(memo.metadata)]
        lines.append("")
    return "\n".join(lines).rstrip("\n") + "\n"

lazytools.report.render_html ¶

render_html(memo: Memo, *, artifacts: ArtifactResolvers | None = None) -> str

Render a :class:Memo to minimal, self-contained HTML; every value is escaped.

Figures are resolved through artifacts (core file:/bytes: registry when omitted) and embedded as base64 data URIs; only image MIME types may be embedded. An unresolvable ref raises rather than rendering a silently incomplete report.

Source code in src/lazytools/report/render.py

def render_html(memo: Memo, *, artifacts: ArtifactResolvers | None = None) -> str:
    """Render a :class:`Memo` to minimal, self-contained HTML; every value is escaped.

    Figures are resolved through ``artifacts`` (core ``file:``/``bytes:``
    registry when omitted) and embedded as base64 data URIs; only image MIME
    types may be embedded. An unresolvable ref raises rather than rendering
    a silently incomplete report.
    """
    if artifacts is None:
        artifacts = ArtifactResolvers()
    parts: list[str] = [
        "<!DOCTYPE html>",
        '<html lang="en">',
        "<head>",
        '<meta charset="utf-8">',
        f"<title>{html.escape(memo.title)}</title>",
        "</head>",
        "<body>",
        f"<h1>{html.escape(memo.title)}</h1>",
    ]
    if memo.as_of is not None:
        parts.append(f'<p class="as-of"><em>as of {html.escape(memo.as_of.isoformat())}</em></p>')
    for section in memo.sections:
        parts.append(f"<h2>{html.escape(section.title)}</h2>")
        for paragraph in section.body.split("\n\n"):
            if paragraph.strip():
                parts.append("<p>" + html.escape(paragraph).replace("\n", "<br>") + "</p>")
        for table in section.tables:
            parts.append("<table>")
            parts.append(
                "<thead><tr>" + "".join(f"<th>{html.escape(cell)}</th>" for cell in table.columns) + "</tr></thead>"
            )
            parts.append("<tbody>")
            for row in table.rows:
                parts.append("<tr>" + "".join(f"<td>{html.escape(cell)}</td>" for cell in row) + "</tr>")
            parts.append("</tbody>")
            parts.append("</table>")
        for figure in section.figures:
            data, mime = artifacts.resolve(figure.ref)
            if not _IMAGE_MIME_RE.match(mime):
                raise ValueError(
                    f"figure {figure.ref!r} resolved to non-image or unsafe MIME {mime!r}; "
                    "only a strict image/* MIME can be embedded in an HTML report"
                )
            src = f"data:{mime};base64,{base64.b64encode(data).decode('ascii')}"
            parts.append("<figure>")
            parts.append(f'<img src="{src}" alt="{html.escape(figure.caption)}">')
            if figure.caption:
                parts.append(f"<figcaption>{html.escape(figure.caption)}</figcaption>")
            parts.append("</figure>")
    if memo.metadata:
        parts.append("<dl>")
        for key in sorted(memo.metadata):
            parts.append(f"<dt>{html.escape(key)}</dt><dd>{html.escape(memo.metadata[key])}</dd>")
        parts.append("</dl>")
    parts += ["</body>", "</html>"]
    return "\n".join(parts) + "\n"

lazytools.report.ArtifactResolvers ¶

ArtifactResolvers(*, file_base_dir: str | None = None)

A per-scheme registry turning artifact refs into (bytes, mime).

file_base_dir optionally sandboxes the file: scheme: when set, refs resolving outside that directory are refused — pass it when the refs come from an untrusted composer (an LLM agent); leave it None for trusted in-process callers.

Source code in src/lazytools/report/artifacts.py

def __init__(self, *, file_base_dir: str | None = None) -> None:
    self._file_base = Path(file_base_dir).resolve() if file_base_dir else None
    self._resolvers: dict[str, Resolver] = {
        "file": self._resolve_file,
        "bytes": self._resolve_bytes,
    }

register ¶

register(scheme: str, resolver: Resolver) -> None

Register (or replace) the resolver for a scheme.

Source code in src/lazytools/report/artifacts.py

def register(self, scheme: str, resolver: Resolver) -> None:
    """Register (or replace) the resolver for a scheme."""
    self._resolvers[scheme] = resolver

schemes ¶

schemes() -> list[str]

The schemes currently resolvable, sorted.

Source code in src/lazytools/report/artifacts.py

def schemes(self) -> list[str]:
    """The schemes currently resolvable, sorted."""
    return sorted(self._resolvers)

resolve ¶

resolve(ref: str) -> tuple[bytes, str]

Resolve a canonical ref to (payload bytes, mime type).

Source code in src/lazytools/report/artifacts.py

def resolve(self, ref: str) -> tuple[bytes, str]:
    """Resolve a canonical ref to ``(payload bytes, mime type)``."""
    scheme, key = split_ref(ref)
    resolver = self._resolvers.get(scheme)
    if resolver is None:
        raise ValueError(
            f"no resolver registered for artifact scheme {scheme!r} "
            f"(ref {ref!r}); available: {', '.join(self.schemes())}"
        )
    return resolver(key)

lazytools.report.ecosystem_resolvers ¶

ecosystem_resolvers(*, regimes_db: Any = None, crawler_db: Any = None, datahub_db_path: str | None = None, file_base_dir: str | None = None) -> ArtifactResolvers

A registry with every ecosystem scheme registered.

regimes: and chart: are always registered (they fail with the install hint only if actually used without their package); crawler: only when crawler_db is given, since it has no session default.

Source code in src/lazytools/report/resolvers.py

def ecosystem_resolvers(
    *,
    regimes_db: Any = None,
    crawler_db: Any = None,
    datahub_db_path: str | None = None,
    file_base_dir: str | None = None,
) -> ArtifactResolvers:
    """A registry with every ecosystem scheme registered.

    ``regimes:`` and ``chart:`` are always registered (they fail with the
    install hint only if actually used without their package); ``crawler:``
    only when ``crawler_db`` is given, since it has no session default.
    """
    resolvers = ArtifactResolvers(file_base_dir=file_base_dir)
    resolvers.register("regimes", regimes_resolver(regimes_db))
    resolvers.register("chart", chart_resolver(datahub_db_path))
    if crawler_db is not None:
        resolvers.register("crawler", crawler_resolver(crawler_db))
    return resolvers

lazytools.report.ReportTools ¶

ReportTools(*, artifacts: ArtifactResolvers | None = None, files: ReportFiles | None = None)

A ToolProvider exposing the deterministic memo renderers.

artifacts configures how figure refs are resolved when rendering HTML; the default registry handles only file: and bytes:. Pass a registry with source resolvers (regimes:, crawler:, chart:) registered — and, since the memo comes from an agent, prefer one constructed with a file_base_dir sandbox.

Source code in src/lazytools/report/tools.py

def __init__(
    self,
    *,
    artifacts: ArtifactResolvers | None = None,
    files: ReportFiles | None = None,
) -> None:
    self._artifacts = artifacts
    self._files = files

lazytools.report.ReportFiles ¶

ReportFiles(*, base_dir: str | PathLike[str] = 'reports')

A ToolProvider exposing save_report (write text → sandboxed file).

Source code in src/lazytools/report/files.py

def __init__(self, *, base_dir: str | os.PathLike[str] = "reports") -> None:
    #: All files are written under here; created on first write.
    self._base = Path(base_dir)

save ¶

save(filename: str, content: str) -> str

Write content to a sandboxed file and return its absolute path.

Public entry point (also used by :class:~lazytools.report.tools.ReportTools's render-and-save tools, which must write large HTML without routing it back through the LLM).

Source code in src/lazytools/report/files.py

def save(self, filename: str, content: str) -> str:
    """Write ``content`` to a sandboxed file and return its absolute path.

    Public entry point (also used by :class:`~lazytools.report.tools.ReportTools`'s
    render-and-save tools, which must write large HTML without routing it
    back through the LLM).
    """
    name = self._safe_name(filename)
    base = self._base.resolve()
    base.mkdir(parents=True, exist_ok=True)
    path = base / name
    # Defence in depth: ``name`` is already a basename, but the target may
    # be a pre-existing symlink pointing outside ``base`` — writing through
    # it would escape the sandbox. Refuse symlinks and any path that does
    # not resolve to a direct child of ``base``.
    if path.is_symlink() or path.resolve().parent != base:
        raise ValueError(f"save_report: refusing to write through a symlink or out-of-sandbox path: {name!r}")
    path.write_text(content, encoding="utf-8")
    return str(path)

Documents¶

lazytools.documents.read_folder_docs ¶

read_folder_docs(path: str, extensions: str = 'txt,md,pdf,docx,html', html_mode: str = 'parsed', recursive: bool = False, output_format: str = 'text', *, base_dir: str | None = None, max_file_bytes: int | None = DEFAULT_MAX_FILE_BYTES, max_files: int | None = DEFAULT_MAX_FILES) -> str

Read documents from a file or folder and return their text content.

Accepts either a single file path or a folder path. When given a folder, scans for all matching files (optionally recursive). When given a file, reads that file directly regardless of the extensions filter.

Supported formats: .txt, .md, .pdf, .docx, .html/.htm. HTML files can be returned as clean extracted body text, raw HTML, or both.

Parameters:

Name	Type	Description	Default
`path`	`str`	Path to a single file OR a folder to scan. File example: "/reports/q4.pdf" Folder example: "/reports"	required
`extensions`	`str`	Comma-separated list of file extensions to include when scanning a folder. Ignored when path points to a single file. Supported values: txt, md, pdf, docx, html. Default: "txt,md,pdf,docx,html" (all formats). Example: "pdf,docx" to read only PDFs and Word files.	`'txt,md,pdf,docx,html'`
`html_mode`	`str`	How to process HTML and HTM files. "parsed" — clean readable text extracted by trafilatura (default). "full" — raw HTML source, unmodified. "both" — parsed body text first, then raw HTML source.	`'parsed'`
`recursive`	`bool`	Whether to search subfolders recursively when path is a folder. False (default) — top-level files only. True — all files in all subfolders. Ignored when path points to a single file.	`False`
`output_format`	`str`	How to format the combined output. "text" (default) — a single human/LLM-readable string with headers. "json" — a JSON object with a "records" array (one entry per file, each with per-file metadata and content) plus truncation fields: "truncated" (bool), "max_files" (the cap applied), and "total_found" (matches discovered before the cap). Parse the output with `json.loads` and index `["records"]` for the file list. Note: the empty-folder case below is reported as a plain string even when "json" is requested, so guard `json.loads` for it (e.g. only parse output that starts with "{").	`'text'`

Returns:

Type	Description
`str`	A single string. For `output_format="text"` this is the concatenated,
`str`	human/LLM-readable document text; for `output_format="json"` it is the
`str`	serialized JSON object described above. One case always returns a plain
`str`	(non-JSON) description string regardless of `output_format`: when a
`str`	scanned folder contains no files matching `extensions`
`str`	("[No documents found ...]").

Raises:

Type	Description
`FileNotFoundError`	When `path` does not exist.
`PermissionError`	When `base_dir` is set and `path` resolves outside it.

Source code in src/lazytools/documents/read_docs.py

def read_folder_docs(
    path: str,
    extensions: str = "txt,md,pdf,docx,html",
    html_mode: str = "parsed",
    recursive: bool = False,
    output_format: str = "text",
    *,
    base_dir: str | None = None,
    max_file_bytes: int | None = DEFAULT_MAX_FILE_BYTES,
    max_files: int | None = DEFAULT_MAX_FILES,
) -> str:
    """Read documents from a file or folder and return their text content.

    Accepts either a single file path or a folder path.
    When given a folder, scans for all matching files (optionally recursive).
    When given a file, reads that file directly regardless of the extensions filter.

    Supported formats: .txt, .md, .pdf, .docx, .html/.htm.
    HTML files can be returned as clean extracted body text, raw HTML, or both.

    Args:
        path: Path to a single file OR a folder to scan.
            File example:   "/reports/q4.pdf"
            Folder example: "/reports"
        extensions: Comma-separated list of file extensions to include when
            scanning a folder. Ignored when path points to a single file.
            Supported values: txt, md, pdf, docx, html.
            Default: "txt,md,pdf,docx,html" (all formats).
            Example: "pdf,docx" to read only PDFs and Word files.
        html_mode: How to process HTML and HTM files.
            "parsed" — clean readable text extracted by trafilatura (default).
            "full"   — raw HTML source, unmodified.
            "both"   — parsed body text first, then raw HTML source.
        recursive: Whether to search subfolders recursively when path is a folder.
            False (default) — top-level files only.
            True  — all files in all subfolders.
            Ignored when path points to a single file.
        output_format: How to format the combined output.
            "text" (default) — a single human/LLM-readable string with headers.
            "json" — a JSON object with a "records" array (one entry per file,
                each with per-file metadata and content) plus truncation
                fields: "truncated" (bool), "max_files" (the cap applied), and
                "total_found" (matches discovered before the cap). Parse the
                output with ``json.loads`` and index ``["records"]`` for the
                file list. Note: the empty-folder case below is reported as a
                plain string even when "json" is requested, so guard
                ``json.loads`` for it (e.g. only parse output that starts
                with "{").

    Returns:
        A single string. For ``output_format="text"`` this is the concatenated,
        human/LLM-readable document text; for ``output_format="json"`` it is the
        serialized JSON object described above. One case always returns a plain
        (non-JSON) description string regardless of ``output_format``: when a
        scanned folder contains no files matching ``extensions``
        ("[No documents found ...]").

    Raises:
        FileNotFoundError: When ``path`` does not exist.
        PermissionError: When ``base_dir`` is set and ``path`` resolves
            outside it.
    """
    target = Path(path).expanduser().resolve()

    # When exposed as an agent tool, `path` is LLM-controlled and therefore
    # untrusted.  If the caller supplies `base_dir`, refuse any path that
    # resolves outside that sandbox.
    if base_dir is not None:
        base = Path(base_dir).expanduser().resolve()
        try:
            target.relative_to(base)
        except ValueError as exc:
            raise PermissionError(f"refused — path {str(target)!r} escapes base_dir {str(base)!r}") from exc

    if not target.exists():
        raise FileNotFoundError(f"path not found — {path}")

    if target.is_file():
        files = [target]
        root = target.parent
    elif target.is_dir():
        root = target
        exts: set[str] = set()
        for e in extensions.split(","):
            e = e.strip().lstrip(".").lower()
            if e:
                exts.add(f".{e}")
        if ".html" in exts:
            exts.add(".htm")
        glob_pattern = "**/*" if recursive else "*"
        # Walk the tree without following symlinks.  Doing so closes
        # symlink-loop hangs and prevents a symlink in the indexed
        # folder from silently widening the read surface to other
        # directories.
        files = sorted(
            f for f in root.glob(glob_pattern) if f.is_file() and not f.is_symlink() and f.suffix.lower() in exts
        )
        if not files:
            return f"[No documents found in '{path}' matching extensions: {extensions}]"
    else:
        raise ValueError(f"path is neither a file nor a directory — {path}")

    # Cap the number of files read in one call so a folder with thousands of
    # documents can't be slurped wholesale into a single tool result.
    files_truncated = False
    total_found = len(files)
    if max_files is not None and len(files) > max_files:
        files = files[:max_files]
        files_truncated = True

    records: list[dict] = []
    for fpath in files:
        suffix = fpath.suffix.lower()
        try:
            size = fpath.stat().st_size
        except OSError as exc:
            # The file may have vanished between glob and stat, or be
            # unreadable. Record the failure and move on rather than aborting
            # the whole scan.
            records.append(
                {
                    "filename": fpath.name,
                    "relative_path": str(fpath.relative_to(root)),
                    "extension": suffix.lstrip("."),
                    "size_bytes": 0,
                    "char_count": 0,
                    "content": f"[Error accessing file: {exc}]",
                }
            )
            continue
        reader = _EXT_READERS.get(suffix)
        if reader is None:
            content = f"[Unsupported extension: {suffix}]"
        elif max_file_bytes is not None and size > max_file_bytes:
            # Bound the on-disk size we will read into memory. This is a first
            # line of defence against memory exhaustion; note a small but
            # heavily-compressed file (e.g. a PDF) can still expand on extract.
            content = f"[Skipped: file is {size:,} bytes, exceeds max_file_bytes={max_file_bytes:,}]"
        else:
            try:
                content = reader(fpath, html_mode)  # type: ignore[operator]
            except Exception as exc:
                content = f"[Error reading file: {exc}]"
        records.append(
            {
                "filename": fpath.name,
                "relative_path": str(fpath.relative_to(root)),
                "extension": suffix.lstrip("."),
                "size_bytes": size,
                "char_count": len(content),
                "content": content,
            }
        )

    if output_format == "json":
        # Wrap in an object so callers can detect truncation. Returning a bare
        # list would silently drop the cap from JSON consumers (the text branch
        # appends a "NOTE" line, but downstream code parses the JSON shape).
        payload = {
            "records": records,
            "truncated": files_truncated,
            "max_files": max_files,
            "total_found": total_found,
        }
        return json.dumps(payload, ensure_ascii=False, indent=2)

    parts: list[str] = []
    for rec in records:
        header = (
            f"{'=' * 72}\n"
            f"FILE : {rec['relative_path']}\n"
            f"TYPE : {rec['extension'].upper()}   SIZE : {rec['size_bytes']:,} bytes   CHARS : {rec['char_count']:,}\n"
            f"{'=' * 72}"
        )
        parts.append(f"{header}\n\n{rec['content']}")

    truncation_note = (
        f" | NOTE: file list truncated to the first {max_files} files" if files_truncated else ""
    )
    summary = (
        f"[{len(records)} document(s) read from '{path}' | "
        f"extensions: {extensions} | html_mode: {html_mode} | recursive: {recursive}{truncation_note}]\n"
        f"{'─' * 72}\n\n"
    )
    return summary + "\n\n".join(parts)

lazytools.documents.read_docs_tools ¶

read_docs_tools(*, base_dir: str, max_file_bytes: int | None = DEFAULT_MAX_FILE_BYTES, max_files: int | None = DEFAULT_MAX_FILES) -> list[Tool]

Return a single-element list with read_folder_docs wrapped as a Tool.

Parameters:

Name	Type	Description	Default
`base_dir`	`str`	Sandbox directory — required. `read_folder_docs` rejects any path that resolves outside this directory at runtime. The tool's `path` argument is LLM-controlled and therefore untrusted; without a sandbox an agent could read arbitrary files on the host (`/etc/passwd`, SSH keys, `.env` files, etc.). Passing `None` (or an empty string) raises `ValueError` — call `read_folder_docs` directly if you genuinely need un-sandboxed access from trusted code.	required
`max_file_bytes`	`int \| None`	Per-file size ceiling; a larger file is reported as skipped instead of read. Defaults to `DEFAULT_MAX_FILE_BYTES`.	`DEFAULT_MAX_FILE_BYTES`
`max_files`	`int \| None`	Ceiling on the number of files read per folder scan. Defaults to `DEFAULT_MAX_FILES`.	`DEFAULT_MAX_FILES`

Source code in src/lazytools/documents/read_docs.py

def read_docs_tools(
    *,
    base_dir: str,
    max_file_bytes: int | None = DEFAULT_MAX_FILE_BYTES,
    max_files: int | None = DEFAULT_MAX_FILES,
) -> list[Tool]:
    """Return a single-element list with ``read_folder_docs`` wrapped as a Tool.

    Args:
        base_dir: Sandbox directory — **required**. ``read_folder_docs`` rejects
            any path that resolves outside this directory at runtime. The tool's
            ``path`` argument is LLM-controlled and therefore untrusted; without
            a sandbox an agent could read arbitrary files on the host
            (``/etc/passwd``, SSH keys, ``.env`` files, etc.). Passing ``None``
            (or an empty string) raises ``ValueError`` — call ``read_folder_docs``
            directly if you genuinely need un-sandboxed access from trusted code.
        max_file_bytes: Per-file size ceiling; a larger file is reported as
            skipped instead of read. Defaults to ``DEFAULT_MAX_FILE_BYTES``.
        max_files: Ceiling on the number of files read per folder scan.
            Defaults to ``DEFAULT_MAX_FILES``.
    """
    from lazybridge import Tool

    if not base_dir:
        raise ValueError(
            "read_docs_tools(base_dir=...) is required. The tool's path argument "
            "is LLM-controlled, so without a sandbox an agent could read ANY file "
            "on the host. Pass base_dir='/safe/directory', or call read_folder_docs "
            "directly for trusted, non-LLM usage."
        )

    def _bound(
        path: str,
        extensions: str = "txt,md,pdf,docx,html",
        html_mode: str = "parsed",
        recursive: bool = False,
        output_format: str = "text",
    ) -> str:
        """Read documents from a file or folder, restricted to base_dir."""
        return read_folder_docs(
            path,
            extensions=extensions,
            html_mode=html_mode,
            recursive=recursive,
            output_format=output_format,
            base_dir=base_dir,
            max_file_bytes=max_file_bytes,
            max_files=max_files,
        )

    return [Tool(_bound, name="read_folder_docs", description=read_folder_docs.__doc__)]

Skills¶

lazytools.skills.build_skill ¶

build_skill(source_dirs: Annotated[list[str], 'One or more folders containing documentation to index.'], skill_name: Annotated[str, 'Skill name — used as the bundle folder name and title.'], output_root: Annotated[str, 'Parent directory for the generated bundle.'] = './generated_skills', description: Annotated[str, 'What this skill covers (used in SKILL.md and tool description).'] = '', usage_notes: Annotated[str, 'Extra operational rules appended to SKILL.md.'] = '', include_extensions: Annotated[list[str], 'File extensions to index.'] = list(DEFAULT_EXTENSIONS), chunk_size: Annotated[int, 'Maximum characters per chunk.'] = 1800, chunk_overlap: Annotated[int, 'Overlap between char-mode chunks.'] = 180, copy_sources: Annotated[bool, 'Copy original docs into the bundle under sources/.'] = False, overwrite: Annotated[bool, 'Replace an existing bundle with the same name.'] = True, max_chars_per_file: Annotated[int, 'Safety cap on characters read per file.'] = 200000) -> dict[str, Any]

Index documentation folders and write a portable skill bundle to disk.

The bundle contains SKILL.md (LLM instructions), manifest.json (metadata + avgdl for BM25), vocab.json (Robertson IDF weights), and chunks.jsonl. Returns a metadata dict: skill_dir, indexed_files, total_chunks, avgdl.

Source code in src/lazytools/skills/doc_skills.py

def build_skill(
    source_dirs: Annotated[list[str], "One or more folders containing documentation to index."],
    skill_name: Annotated[str, "Skill name — used as the bundle folder name and title."],
    output_root: Annotated[str, "Parent directory for the generated bundle."] = "./generated_skills",
    description: Annotated[str, "What this skill covers (used in SKILL.md and tool description)."] = "",
    usage_notes: Annotated[str, "Extra operational rules appended to SKILL.md."] = "",
    include_extensions: Annotated[list[str], "File extensions to index."] = list(DEFAULT_EXTENSIONS),  # noqa: B006
    chunk_size: Annotated[int, "Maximum characters per chunk."] = 1800,
    chunk_overlap: Annotated[int, "Overlap between char-mode chunks."] = 180,
    copy_sources: Annotated[bool, "Copy original docs into the bundle under sources/."] = False,
    overwrite: Annotated[bool, "Replace an existing bundle with the same name."] = True,
    max_chars_per_file: Annotated[int, "Safety cap on characters read per file."] = 200_000,
) -> dict[str, Any]:
    """
    Index documentation folders and write a portable skill bundle to disk.

    The bundle contains SKILL.md (LLM instructions), manifest.json (metadata +
    avgdl for BM25), vocab.json (Robertson IDF weights), and chunks.jsonl.
    Returns a metadata dict: skill_dir, indexed_files, total_chunks, avgdl.
    """
    roots = [Path(p).expanduser().resolve() for p in source_dirs]
    for root in roots:
        if not root.is_dir():
            raise FileNotFoundError(f"Not a directory: {root}")

    skill_dir = Path(output_root).expanduser().resolve() / _slugify(skill_name)
    if skill_dir.exists():
        if not overwrite:
            raise FileExistsError(f"Skill already exists: {skill_dir}")
        # Only ever delete something that is recognisably a skill bundle.
        # ``overwrite=True`` must not become an arbitrary recursive delete
        # when output_root/skill_name happens to collide with an unrelated
        # directory.
        if not (skill_dir / "manifest.json").exists():
            raise FileExistsError(
                f"Refusing to overwrite {skill_dir}: the directory exists but does not "
                f"look like a skill bundle (no manifest.json). Remove it manually if "
                f"replacing it is intended."
            )
        shutil.rmtree(skill_dir)
    skill_dir.mkdir(parents=True)
    if copy_sources:
        (skill_dir / "sources").mkdir()

    indexed_files: list[str] = []
    all_chunks: list[DocChunk] = []

    for root in roots:
        for path in _iter_docs([root], include_extensions):
            try:
                text = path.read_text(encoding="utf-8", errors="ignore")[:max_chars_per_file]
            except Exception:
                continue
            chunks = _make_chunks(path, text, chunk_size, chunk_overlap)
            if not chunks:
                continue
            indexed_files.append(str(path))
            all_chunks.extend(chunks)
            if copy_sources:
                dest = skill_dir / "sources" / root.name / path.parent.relative_to(root)
                dest.mkdir(parents=True, exist_ok=True)
                shutil.copy2(path, dest / path.name)

    if not all_chunks:
        raise ValueError("No indexable documentation found in the provided folders.")

    description = description or f"Documentation skill built from {len(indexed_files)} files."
    avgdl = sum(c.doc_len for c in all_chunks) / len(all_chunks)
    idf = _build_idf(all_chunks)

    manifest = SkillManifest(
        name=skill_name,
        description=description,
        source_dirs=[str(p) for p in roots],
        indexed_files=indexed_files,
        total_chunks=len(all_chunks),
        avgdl=avgdl,
        extensions=list(include_extensions),
    )
    skill_md = _render_skill_md(
        name=skill_name,
        description=description,
        usage_notes=usage_notes,
        file_count=len(indexed_files),
        total_chunks=len(all_chunks),
        source_dirs=roots,
    )

    (skill_dir / "SKILL.md").write_text(skill_md, encoding="utf-8")
    (skill_dir / "manifest.json").write_text(
        json.dumps(asdict(manifest), ensure_ascii=False, indent=2), encoding="utf-8"
    )
    (skill_dir / "vocab.json").write_text(json.dumps(idf, ensure_ascii=False, indent=2), encoding="utf-8")
    with (skill_dir / "chunks.jsonl").open("w", encoding="utf-8") as f:
        for chunk in all_chunks:
            f.write(json.dumps(asdict(chunk), ensure_ascii=False) + "\n")

    return {
        "skill_dir": str(skill_dir),
        "skill_name": skill_name,
        "description": description,
        "indexed_files": indexed_files,
        "total_chunks": len(all_chunks),
        "avgdl": round(avgdl, 1),
    }

lazytools.skills.query_skill ¶

query_skill(skill_dir: Annotated[str, 'Path to a skill bundle created by build_skill().'], task: Annotated[str, 'Question or task to answer from the indexed documentation.'], mode: Annotated[Literal['auto', 'answer', 'extract', 'locate', 'summarize'], "Execution mode. 'auto' detects intent from the task wording."] = 'auto', top_k: Annotated[int, 'Number of chunks to retrieve.'] = 8, max_chars: Annotated[int, 'Maximum characters in the returned context brief.'] = 10000, include_quotes: Annotated[bool, 'Append full excerpts after evidence bullets.'] = True) -> str

Retrieve the most relevant chunks via BM25 and return a grounded context brief ready to be injected into an LLM's context window.

Source code in src/lazytools/skills/doc_skills.py

def query_skill(
    skill_dir: Annotated[str, "Path to a skill bundle created by build_skill()."],
    task: Annotated[str, "Question or task to answer from the indexed documentation."],
    mode: Annotated[
        Literal["auto", "answer", "extract", "locate", "summarize"],
        "Execution mode. 'auto' detects intent from the task wording.",
    ] = "auto",
    top_k: Annotated[int, "Number of chunks to retrieve."] = 8,
    max_chars: Annotated[int, "Maximum characters in the returned context brief."] = 10_000,
    include_quotes: Annotated[bool, "Append full excerpts after evidence bullets."] = True,
) -> str:
    """
    Retrieve the most relevant chunks via BM25 and return a grounded context
    brief ready to be injected into an LLM's context window.
    """
    sdir = Path(skill_dir).expanduser().resolve()
    if not sdir.exists():
        raise FileNotFoundError(f"Skill directory not found: {sdir}")

    manifest = _load_manifest(sdir)
    chunks = _load_chunks(sdir)
    idf = _load_idf(sdir)
    skill_md = _load_skill_md(sdir)
    resolved_mode: MODE = _auto_mode(task) if mode == "auto" else mode  # type: ignore[assignment]

    q_tokens = _tokenize(task)
    ranked = sorted(((c, _bm25(c, q_tokens, idf, manifest.avgdl)) for c in chunks), key=lambda x: x[1], reverse=True)
    selected = [c for c, score in ranked[: max(1, top_k)] if score > 0]

    if not selected:
        return (
            f"[skill] {manifest.name}\n[task] {task}\n\n"
            "No relevant documentation was retrieved for this task. "
            "Do not answer beyond the indexed evidence."
        )[:max_chars]

    brief = _build_brief(manifest=manifest, skill_md=skill_md, task=task, mode=resolved_mode, selected=selected)
    result_lines: list[str] = []

    if resolved_mode == "locate":
        result_lines = ["Relevant files:"] + [f"  • {p}" for p in dict.fromkeys(c.path for c in selected)]
    elif resolved_mode == "extract":
        result_lines = ["Excerpts:"]
        for c in selected:
            result_lines.append(f"\n### {c.heading}  [{Path(c.path).name}]\n{_trim(c.text, 1000)}")
    elif resolved_mode == "summarize":
        result_lines = ["Summary:"]
        for c in selected[:6]:
            condensed = re.sub(r"\s+", " ", c.text)
            result_lines.append(f"  - {_trim(condensed, 300)}  [{Path(c.path).name}]")
    else:
        result_lines = ["Best evidence:"]
        for c in selected[:5]:
            condensed = re.sub(r"\s+", " ", c.text)
            result_lines.append(f"  - {_trim(condensed, 400)}  [{Path(c.path).name}]")
        if include_quotes:
            result_lines.append("\nFull excerpts:")
            for c in selected[:3]:
                result_lines.append(f"\n### {c.heading}  [{Path(c.path).name}]\n{_trim(c.text, 800)}")

    return (brief + "\n\n[result]\n" + "\n".join(result_lines))[:max_chars]

lazytools.skills.skill_tools ¶

skill_tools(*, skill_dir: Annotated[str, 'Path to a skill bundle created by build_skill().'], name: Annotated[str | None, 'Tool name exposed to the agent.'] = None, description: Annotated[str | None, 'Tool description.'] = None, strict: Annotated[bool, 'Strict JSON schema validation.'] = False) -> list[Tool]

Return a single-element list containing a query_skill() Tool ready to be passed to any agent or pipeline.

Source code in src/lazytools/skills/doc_skills.py

def skill_tools(
    *,
    skill_dir: Annotated[str, "Path to a skill bundle created by build_skill()."],
    name: Annotated[str | None, "Tool name exposed to the agent."] = None,
    description: Annotated[str | None, "Tool description."] = None,
    strict: Annotated[bool, "Strict JSON schema validation."] = False,
) -> list[Tool]:
    """Return a single-element list containing a query_skill() Tool ready
    to be passed to any agent or pipeline."""
    sdir = Path(skill_dir).expanduser().resolve()
    manifest = _load_manifest(sdir)

    def _run(
        task: Annotated[str, "Question or task to answer from this skill."],
        mode: Annotated[Literal["auto", "answer", "extract", "locate", "summarize"], "Retrieval mode."] = "auto",
        top_k: Annotated[int, "Number of chunks to retrieve."] = 8,
        include_quotes: Annotated[bool, "Include full excerpts."] = True,
    ) -> str:
        """Query a local documentation skill and return a grounded context brief.

        Use when the task is about the documentation indexed by this skill;
        treat the result as grounded evidence and answer only from it.
        """
        return query_skill(str(sdir), task, mode=mode, top_k=top_k, include_quotes=include_quotes)

    return [
        Tool(
            _run,
            name=name or _slugify(manifest.name),
            description=description or manifest.description,
            strict=strict,
        )
    ]

lazytools.skills.skill_builder_tools ¶

skill_builder_tools(*, base_dir: Annotated[str, 'Sandbox directory — required. Source dirs must resolve inside it; bundles are written to <base_dir>/generated_skills.'], name: Annotated[str, 'Tool name.'] = 'build_doc_skill', description: Annotated[str, 'Tool description.'] = 'Index documentation folders into a reusable local skill bundle. Call this to transform one or more documentation folders into a queryable local skill.', strict: Annotated[bool, 'Strict JSON schema validation.'] = False) -> list[Tool]

Return a single-element list containing a Tool that builds skill bundles.

Parameters:

Name	Type	Description	Default
`base_dir`	`Annotated[str, 'Sandbox directory — required. Source dirs must resolve inside it; bundles are written to <base_dir>/generated_skills.']`	Sandbox directory — required. The tool's `source_dirs` argument is LLM-controlled; without a sandbox an agent could index (and thus read, via `query_skill`) arbitrary files on the host. Every source dir must resolve inside `base_dir`, and bundles are always written under `<base_dir>/generated_skills` — the LLM cannot choose the output location. Call :func:`build_skill` directly from trusted code if you genuinely need an unsandboxed build.	required

Source code in src/lazytools/skills/doc_skills.py

def skill_builder_tools(
    *,
    base_dir: Annotated[str, "Sandbox directory — required. Source dirs must resolve inside it; bundles are written to <base_dir>/generated_skills."],
    name: Annotated[str, "Tool name."] = "build_doc_skill",
    description: Annotated[str, "Tool description."] = (
        "Index documentation folders into a reusable local skill bundle. "
        "Call this to transform one or more documentation folders into a queryable local skill."
    ),
    strict: Annotated[bool, "Strict JSON schema validation."] = False,
) -> list[Tool]:
    """Return a single-element list containing a Tool that builds skill bundles.

    Args:
        base_dir: Sandbox directory — **required**. The tool's ``source_dirs``
            argument is LLM-controlled; without a sandbox an agent could index
            (and thus read, via ``query_skill``) arbitrary files on the host.
            Every source dir must resolve inside ``base_dir``, and bundles are
            always written under ``<base_dir>/generated_skills`` — the LLM
            cannot choose the output location. Call :func:`build_skill`
            directly from trusted code if you genuinely need an unsandboxed
            build.
    """
    if not base_dir:
        raise ValueError(
            "skill_builder_tools(base_dir=...) is required. The tool's source_dirs "
            "argument is LLM-controlled, so without a sandbox an agent could index "
            "and read ANY file on the host. Pass base_dir='/safe/directory', or call "
            "build_skill directly for trusted, non-LLM usage."
        )
    base = Path(base_dir).expanduser().resolve()
    output_root = base / "generated_skills"

    def _bound(
        source_dirs: Annotated[list[str], "Documentation folders to index. Must be inside the sandbox directory."],
        skill_name: Annotated[str, "Skill name — used as the bundle folder name and title."],
        description: Annotated[str, "What this skill covers (used in SKILL.md and tool description)."] = "",
        usage_notes: Annotated[str, "Extra operational rules appended to SKILL.md."] = "",
    ) -> dict[str, Any]:
        """Index documentation folders into a skill bundle, restricted to the sandbox."""
        for d in source_dirs:
            resolved = Path(d).expanduser().resolve()
            try:
                resolved.relative_to(base)
            except ValueError as exc:
                raise PermissionError(
                    f"refused — source dir {str(resolved)!r} escapes base_dir {str(base)!r}"
                ) from exc
        return build_skill(
            source_dirs,
            skill_name,
            output_root=str(output_root),
            description=description,
            usage_notes=usage_notes,
        )

    return [Tool(_bound, name=name, description=description, strict=strict)]

lazytools.skills.skill_pipeline ¶

skill_pipeline(*, skill_dir: Annotated[str, 'Path to a skill bundle.'], provider: Annotated[str | Any, 'LazyBridge provider alias or instance.'] = 'anthropic', router_model: Annotated[str | None, 'Model for the task-sharpening router.'] = None, executor_model: Annotated[str | None, 'Model for the grounded-answer executor.'] = None, session: Annotated[Any, 'Optional Session. Created if omitted.'] = None, native_tools: Annotated[list | None, 'Provider-native tools for the executor.'] = None) -> Tool

Two-step pipeline exposed as a single Tool.

Router — rewrites the user task into a retrieval-optimised query.
Executor — calls skill_tools() and synthesises a grounded answer.

Returns an Agent.chain(router, executor).as_tool().

Source code in src/lazytools/skills/doc_skills.py

def skill_pipeline(
    *,
    skill_dir: Annotated[str, "Path to a skill bundle."],
    provider: Annotated[str | Any, "LazyBridge provider alias or instance."] = "anthropic",
    router_model: Annotated[str | None, "Model for the task-sharpening router."] = None,
    executor_model: Annotated[str | None, "Model for the grounded-answer executor."] = None,
    session: Annotated[Any, "Optional Session. Created if omitted."] = None,
    native_tools: Annotated[list | None, "Provider-native tools for the executor."] = None,
) -> Tool:
    """
    Two-step pipeline exposed as a single Tool.

      1. Router   — rewrites the user task into a retrieval-optimised query.
      2. Executor — calls skill_tools() and synthesises a grounded answer.

    Returns an Agent.chain(router, executor).as_tool().
    """
    from lazybridge import LLMEngine

    sdir = Path(skill_dir).expanduser().resolve()
    manifest = _load_manifest(sdir)
    sess = session or Session()
    s_tool = skill_tools(skill_dir=str(sdir))[0]

    # Resolve model strings: use the explicit model override if given, else the provider alias.
    router_model_str = router_model or (provider if isinstance(provider, str) else "anthropic")
    executor_model_str = executor_model or (provider if isinstance(provider, str) else "anthropic")

    router = Agent(
        engine=LLMEngine(
            router_model_str,
            system=(
                "You sharpen user queries for a local documentation retrieval system. "
                "Return a single concise retrieval query. "
                "Preserve every technical identifier: class names, method names, "
                "parameter names, error codes, configuration keys. "
                "Do not answer the question. Do not add facts."
            ),
        ),
        name="skill_router",
        session=sess,
    )
    executor = Agent(
        engine=LLMEngine(
            executor_model_str,
            system=(
                "You answer from the local skill tool only. Always call the skill tool first. "
                "Build your answer exclusively from the tool result. "
                "Name every source file you use. "
                "If the skill returns weak or absent evidence, say so explicitly."
            ),
            native_tools=native_tools,
        ),
        name="skill_executor",
        session=sess,
        tools=[s_tool],
    )

    pipeline = Agent.chain(router, executor, name="doc_skill_pipeline", session=sess)
    return pipeline.as_tool(
        "doc_skill_pipeline",
        description=(
            f"Grounded local-docs pipeline: {manifest.description}. "
            "Use for questions grounded in the indexed documentation. "
            "The pipeline sharpens the query then retrieves and synthesises the answer."
        ),
    )

Planners¶

These ship in the LazyBridge core (lazybridge.ext.planners), not in lazytoolkit — documented here for completeness. See the Planners guide. orchestrator_agent / blackboard_orchestrator_agent are the canonical names; the make_* symbols below are the same callables.

The blackboard planner (make_blackboard_planner / blackboard_orchestrator_agent) takes the same arguments — see the Blackboard guide for its full reference.

lazybridge.ext.planners.make_planner ¶

make_planner(agents: list[Agent], *, model: str = 'claude-opus-4-7', system: str | None = None, name: str = 'planner', verbose: bool = False, verify: Agent | None = None, max_verify: int = 3) -> Agent

Build a planner :class:Agent over the given sub-agents.

The returned agent has

each sub-agent in agents as a direct tool (so it can call one when that's enough);
five builder tools (create_plan, add_step, inspect_plan, run_plan, discard_plan) that compose a :class:Plan one step at a time, with local validation per step.

Parameters:

Name	Type	Description	Default
`agents`	`list[Agent]`	The sub-agents the planner may dispatch to. Each must have a unique `.name`; the planner addresses them by that name in `StepSpec.agent`.	required
`model`	`str`	Provider model id for the planner LLM. Default `"claude-opus-4-7"`.	`'claude-opus-4-7'`
`system`	`str \| None`	Override the planner's system prompt. By default we prepend "You are a generalist assistant." to :data:`PLANNER_GUIDANCE` so the LLM has decision rules and worked examples for `execute_plan`.	`None`
`name`	`str`	Display name for the planner agent.	`'planner'`
`verbose`	`bool`	If True, print event traces to stdout.	`False`
`verify`	`Agent \| None`	Optional judge :class:`Agent` that vets the planner's final output. When set, the planner's response runs through `verify` (LazyBridge's built-in verify-with-retry loop). The judge should reply "approved" or "rejected: "; on rejection the planner retries up to `max_verify` times with the judge's feedback in context. Costs one extra LLM call per attempt — use it for tasks where wrong answers are expensive.	`None`
`max_verify`	`int`	Max judge attempts when `verify` is set. Default 3.	`3`

Returns:

Type	Description
`Agent`	A configured planner :class:`Agent`. Call it with the user task.

Raises:

Type	Description
`ValueError`	if `agents` is empty or contains duplicate names.

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

def make_planner(
    agents: list[Agent],
    *,
    model: str = "claude-opus-4-7",
    system: str | None = None,
    name: str = "planner",
    verbose: bool = False,
    verify: Agent | None = None,
    max_verify: int = 3,
) -> Agent:
    """Build a planner :class:`Agent` over the given sub-agents.

    The returned agent has:
      - each sub-agent in ``agents`` as a direct tool (so it can call one
        when that's enough);
      - five **builder** tools (``create_plan``, ``add_step``,
        ``inspect_plan``, ``run_plan``, ``discard_plan``) that compose a
        :class:`Plan` one step at a time, with local validation per step.

    Args:
        agents: The sub-agents the planner may dispatch to. Each must have
            a unique ``.name``; the planner addresses them by that name in
            ``StepSpec.agent``.
        model: Provider model id for the planner LLM. Default
            ``"claude-opus-4-7"``.
        system: Override the planner's system prompt. By default we prepend
            "You are a generalist assistant." to :data:`PLANNER_GUIDANCE`
            so the LLM has decision rules and worked examples for
            ``execute_plan``.
        name: Display name for the planner agent.
        verbose: If True, print event traces to stdout.
        verify: Optional judge :class:`Agent` that vets the planner's final
            output. When set, the planner's response runs through
            ``verify`` (LazyBridge's built-in verify-with-retry loop). The
            judge should reply "approved" or "rejected: <reason>"; on
            rejection the planner retries up to ``max_verify`` times with
            the judge's feedback in context. Costs one extra LLM call per
            attempt — use it for tasks where wrong answers are expensive.
        max_verify: Max judge attempts when ``verify`` is set. Default 3.

    Returns:
        A configured planner :class:`Agent`. Call it with the user task.

    Raises:
        ValueError: if ``agents`` is empty or contains duplicate names.
    """
    if not agents:
        raise ValueError("make_planner: agents list must not be empty")
    names = [a.name for a in agents]
    if len(set(names)) != len(names):
        raise ValueError(f"make_planner: agents must have unique names; got {names}")

    registry = {a.name: a for a in agents}
    builder_tools = make_plan_builder_tools(registry)

    if system is None:
        system = "You are a generalist assistant.\n\n" + PLANNER_GUIDANCE

    return Agent(
        engine=LLMEngine(model, system=system),
        tools=[*agents, *builder_tools],
        name=name,
        verbose=verbose,
        verify=verify,
        max_verify=max_verify,
    )

lazybridge.ext.planners.make_plan_builder_tools ¶

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

Five builder tools that share state via closure.

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

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

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

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

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

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

    plans: dict[str, _PlanInProgress] = {}

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

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

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

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

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

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

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

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

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

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

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

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

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

lazybridge.ext.planners.PlanSpec ¶

Bases: BaseModel

The argument shape of execute_plan.

lazybridge.ext.planners.StepSpec ¶

Bases: BaseModel

One node in the plan DAG.

Composition sugar¶

chain and parallel are Agent classmethods in the LazyBridge core. See the Composition sugar guide.

lazybridge.Agent.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)

lazybridge.Agent.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,
    )

Human-in-the-loop¶

These ship in lazybridge.ext.hil. The *_agent factories return an Agent; HumanEngine / SupervisorEngine are Engines you pass via Agent(engine=…). See the Human-in-the-loop guide.

lazybridge.ext.hil.human_agent ¶

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

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

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

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

from lazybridge.ext.hil import human_agent

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

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

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

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

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

        from lazybridge.ext.hil import human_agent

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

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

lazybridge.ext.hil.supervisor_agent ¶

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

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

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

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

from lazybridge.ext.hil import supervisor_agent

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

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

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

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

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

        from lazybridge.ext.hil import supervisor_agent

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

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

lazybridge.ext.hil.HumanEngine ¶

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

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

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

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

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

lazybridge.ext.hil.SupervisorEngine ¶

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

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

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

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

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

API reference¶

Safety¶

lazytools.safety.Allowlist ¶

lazytools.safety.ConfirmationGate ¶

grant ¶

grant_any ¶

consume ¶

lazytools.safety.ActionBlocked ¶

lazytools.safety.current_scope ¶

lazytools.safety.UrlBlocked ¶

lazytools.safety.validate_public_url ¶

Artifact registry¶

lazytools.registry.DBEntry dataclass ¶

lazytools.registry.KNOWN_DBS module-attribute ¶

lazytools.registry.resolve_db ¶

lazytools.registry.status ¶

lazytools.registry.artifact_dbs ¶

lazytools.registry.register_artifact ¶

lazytools.registry.search_artifacts ¶

lazytools.registry.get_artifact ¶

lazytools.registry.search_everywhere ¶

lazytools.registry.get_everywhere ¶

lazytools.registry.RegistryTools ¶

Gmail¶

lazytools.connectors.gmail.GmailTools ¶

require_confirmation property ¶

confirm_once ¶

confirm_send ¶

lazytools.connectors.gmail.GmailClient ¶

from_credentials classmethod ¶

get_history_id ¶

list_history_message_ids ¶

watch ¶

stop_watch ¶

lazytools.connectors.gmail.parse_authentication_results ¶

Outlook¶

lazytools.connectors.outlook.OutlookTools ¶

require_confirmation property ¶

confirm_once ¶

confirm_send ¶

lazytools.connectors.outlook.OutlookClient ¶

connect classmethod ¶

close ¶

list_message_ids ¶

Telegram¶

lazytools.connectors.telegram.TelegramTools ¶

require_confirmation property ¶

confirm_once ¶

confirm_send ¶

lazytools.connectors.telegram.TelegramClient ¶

from_token classmethod ¶

close ¶

send_document ¶

MCP¶

lazytools.connectors.mcp.MCP ¶

stdio classmethod ¶

http classmethod ¶

from_transport classmethod ¶

lazytools.connectors.mcp.MCPServer ¶

aconnect async ¶

alist_tools async ¶

invalidate_tools_cache ¶

aclose async ¶

as_tools ¶

Code Support Agent¶

lazytools.connectors.code_support.claude_code ¶

Parameters¶

Notes¶

lazytools.connectors.code_support.claude_code_mcp ¶

Parameters¶

lazytools.connectors.code_support.codex ¶

Parameters¶

lazytools.connectors.code_support.codex_mcp ¶

Parameters¶

lazytools.connectors.code_support.codex_reviewer ¶

lazytools.connectors.code_support.codex_consultant ¶

lazytools.connectors.code_support.codex_native_reviewer ¶

lazytools.connectors.code_support.build_cli_collaboration ¶

Parameters¶

Notes¶

lazytools.registry.DBEntry `dataclass` ¶

lazytools.registry.KNOWN_DBS `module-attribute` ¶

require_confirmation `property` ¶

from_credentials `classmethod` ¶

require_confirmation `property` ¶

connect `classmethod` ¶

require_confirmation `property` ¶

from_token `classmethod` ¶

stdio `classmethod` ¶

http `classmethod` ¶

from_transport `classmethod` ¶

aconnect `async` ¶

alist_tools `async` ¶

aclose `async` ¶

lazytools.connectors.gateway.ExternalToolSpec `dataclass` ¶

from_mapping `classmethod` ¶

lazybridge.Agent.chain `classmethod` ¶

lazybridge.Agent.parallel `classmethod` ¶

run_branches `async` ¶

run `async` ¶