teff.tool.builtin

teff.tool.builtin

Modules:

Name	Description
calculator	Calculator tool — AST-based safe evaluation of math expressions.
csv	CSV tools — read, filter and aggregate tabular data from CSV files.
data	Data tools — JSON/YAML parsing, a persistent key-value store, and safe eval.
file	File tools — read, write, and edit files on disk.
fs	Filesystem/env tools — list, glob, env vars, and time.
git	Git tools — read-only inspection of a git repository.
github	GitHub tools — list pull requests, fetch diffs, post comments, approve.
gitlab	GitLab tools — list merge requests, fetch diffs, post notes, approve.
http	HTTP tool — send arbitrary HTTP requests to APIs.
human	AskHuman — pause a ReAct run and wait for an operator's answer.
lock	Distributed lock over a Redis-compatible store (Redis/KeyDB/Valkey).
notify	Notification tools — send email via SMTP and Telegram bot messages.
pdf	PDF read tool — extract text from a PDF file.
rag_ingest	Write tool — add documents to a vector store from a workflow YAML.
redis	Redis-like store tools — Redis, KeyDB, Valkey (any RESP server).
s3	S3 tools — list, read, and write objects in Amazon S3 (or S3-compatible stores).
shell	Shell tool — run shell commands asynchronously, without a shell.
slack	Slack tool — send messages to a Slack channel.
sql	SQL tools — read-only queries and schema inspection for SQLite/PostgreSQL.
wait_for	WaitForTool — poll until a condition holds or a timeout elapses.
web_fetch	Web fetch tool — download a URL and extract its text content.
web_search	Web search tool — DuckDuckGo search without API key.

Classes:

Name	Description
AskHuman	Pause the workflow and ask the human operator a question.
CalculatorTool	Evaluate mathematical expressions using AST-based safe eval.
CsvQueryTool	Read, filter and aggregate a CSV file.
CurrentTimeTool	Get the current date and time.
EditFileTool	Edit a file by replacing the first occurrence of a string.
GetEnvTool	Read an environment variable, masking secret-looking values.
GitHubApproveTool	Approve a pull request by submitting an APPROVE review.
GitHubGetPRChangesTool	Fetch a pull request's diff (changed files + per-file patches).
GitHubListOpenPRsTool	List open pull requests for an owner/repo.
GitHubPostCommentTool	Post a comment on a pull request (as an issue comment).
GitLabApproveTool	Approve a merge request.
GitLabGetMRChangesTool	Fetch a merge request's diff (changed files + line-level changes).
GitLabListOpenMRsTool	List open merge requests for a project.
GitLabPostNoteTool	Post a note (comment) on a merge request.
GitTool	Inspect a git repository without mutating it.
GlobTool	Find files matching a glob pattern.
HttpRequestTool	Send an HTTP request to an API endpoint.
JsonParseTool	Parse a JSON string and pretty-print it (or validate it).
KVStoreTool	A persistent JSON-backed key-value store.
ListDirTool	List the contents of a directory.
LockTool	Distributed lock over Redis (KeyDB/Valkey supported).
MemoryTool	Tool that lets an agent read and write long-term memory.
PDFReadTool	Extract text from a PDF file.
PythonEvalTool	Safely evaluate a Python expression using an AST whitelist.
RAGIngestTool	Add documents to a vector store.
ReadFileTool	Read a file's contents.
RedisTool	Read/write a Redis-compatible store (KeyDB/Valkey supported).
S3GetTool	Download an object from an S3 bucket and return its contents.
S3PutTool	Upload text content to an object in an S3 bucket.
S3Tool	List objects in an S3 bucket.
SQLDescribeTool	Describe a table's columns and types.
SQLListTablesTool	List the tables in a database.
SQLQueryTool	Run a read-only SQL query against a database.
SendEmailTool	Send an email message via SMTP.
SendTelegramTool	Send a message via the Telegram Bot API.
ShellTool	Run shell commands without a shell.
SlackSendTool	Send a message to a Slack channel.
WaitForTool	Poll until a condition holds or a timeout elapses.
WebFetchTool	Fetch a URL and return the page's text content.
WebSearchTool	Search the web using DuckDuckGo (no API key required).
WriteFileTool	Write text content to a file.
YamlParseTool	Parse a YAML string and dump it as pretty JSON.

AskHuman

Bases: Tool

Pause the workflow and ask the human operator a question.

Parameters:

Name	Type	Description	Default
question		The question to ask the operator. The run pauses and the operator's free-form reply is returned as the tool result.	required

Requires a checkpointer on graph.run() (the pause is a GraphInterrupt); resume with the answer under the ask_human state key::

graph.run(state, checkpointer=cp, resume={"ask_human": "42"})

Source code in teff/tool/builtin/human.py

class AskHuman(Tool):
    """Pause the workflow and ask the human operator a question.

    Args:
        question: The question to ask the operator.  The run pauses and
            the operator's free-form reply is returned as the tool result.

    Requires a checkpointer on ``graph.run()`` (the pause is a
    ``GraphInterrupt``); resume with the answer under the ``ask_human``
    state key::

        graph.run(state, checkpointer=cp, resume={"ask_human": "42"})
    """

    name = "ask_human"
    description = (
        "Pause the workflow and ask the human operator a question; "
        "returns the operator's free-form reply.  Use only when you truly "
        "need information only a person can provide (a preference, a "
        "decision, or data you cannot compute or fetch yourself)."
    )
    schema = {
        "type": "object",
        "properties": {
            "question": {
                "type": "string",
                "description": "The question to ask the operator.",
            }
        },
        "required": ["question"],
    }

    async def arun(  # type: ignore[override]
        self, question: str = ""
    ) -> str:
        raise NotImplementedError(
            "ask_human is intercepted by the ToolExec node (it pauses the "
            "run as an interrupt); it cannot be invoked directly"
        )

CalculatorTool

Bases: Tool

Evaluate mathematical expressions using AST-based safe eval.

Source code in teff/tool/builtin/calculator.py

class CalculatorTool(Tool):
    """Evaluate mathematical expressions using AST-based safe eval."""

    name = "calculator"
    description = "Evaluate mathematical expressions"

    def run(self, expression: str = "") -> str:  # type: ignore[override]
        tree = ast.parse(expression, mode="eval")
        return str(self._eval(tree.body))

    def _eval(self, node):
        if isinstance(node, ast.Constant):
            return node.value
        if isinstance(node, ast.UnaryOp):
            return _OPS[type(node.op)](self._eval(node.operand))
        if isinstance(node, ast.BinOp):
            return _OPS[type(node.op)](self._eval(node.left), self._eval(node.right))
        if isinstance(node, ast.Name) and node.id == "pi":
            import math

            return math.pi
        raise ValueError(f"unsupported: {ast.dump(node)}")

CsvQueryTool

Bases: Tool

Read, filter and aggregate a CSV file.

Parameters:

Name	Type	Description	Default
action		read | columns | filter | aggregate.	required
path		CSV file path (falls back to config path).	required
column		Column name for filter/aggregate.	required
value		Exact value to match for filter.	required
op		count | sum | avg | min | max.	required
group_by		Optional column to group aggregate by.	required
limit		Max rows to return (default 100).	required

Args (config): path: Default CSV file path.

Source code in teff/tool/builtin/csv.py

class CsvQueryTool(Tool):
    """Read, filter and aggregate a CSV file.

    Args:
        action: ``read`` | ``columns`` | ``filter`` | ``aggregate``.
        path: CSV file path (falls back to config ``path``).
        column: Column name for ``filter``/``aggregate``.
        value: Exact value to match for ``filter``.
        op: ``count`` | ``sum`` | ``avg`` | ``min`` | ``max``.
        group_by: Optional column to group ``aggregate`` by.
        limit: Max rows to return (default 100).

    Args (config):
        path: Default CSV file path.
    """

    name = "csv_query"
    description = (
        "Read, filter and aggregate a CSV file (read, columns, filter, aggregate)"
    )

    def __init__(self, config: dict | None = None):
        cfg = config or {}
        self.path = cfg.get("path", "")

    def _rows(self, path: str):
        path = path or self.path
        if not path:
            raise ValueError("path is required")
        try:
            with open(path, newline="", encoding="utf-8") as f:
                reader = csv.DictReader(f)
                if reader.fieldnames is None:
                    raise ValueError(f"no header row in {path}")
                return reader.fieldnames, [dict(r) for r in reader]
        except FileNotFoundError as e:
            raise ValueError(f"file not found: {path}") from e

    @staticmethod
    def _fmt(v) -> str:
        if isinstance(v, float) and v.is_integer():
            return str(int(v))
        return f"{v:.4f}".rstrip("0").rstrip(".")

    def _render(self, fields, rows, limit: int) -> str:
        out = ["\t".join(fields)]
        for r in rows[:limit]:
            out.append("\t".join(str(r.get(c, "")) for c in fields))
        return "\n".join(out)

    def run(  # type: ignore[override]
        self,
        action: str,
        path: str = "",
        column: str = "",
        value: str = "",
        op: str = "count",
        group_by: str = "",
        limit: int = 100,
    ) -> str:
        if not action:
            raise ValueError("action is required (read, columns, filter, aggregate)")
        fields, rows = self._rows(path)
        a = action.lower()
        if a == "columns":
            return "\n".join(fields)
        if a == "read":
            out = self._render(fields, rows, int(limit))
            return out if len(out) > 1 else f"no rows (columns: {', '.join(fields)})"
        if a == "filter":
            if not column:
                raise ValueError("column is required")
            needle = str(value)
            matches = [r for r in rows if str(r.get(column, "")) == needle]
            if not matches:
                return f"no rows match {column}={needle}"
            return self._render(fields, matches, int(limit))
        if a == "aggregate":
            if not column:
                raise ValueError("column is required")
            groups: dict[str, list] = {}
            for r in rows:
                gkey = str(r.get(group_by, "")) if group_by else ""
                groups.setdefault(gkey, []).append(r)
            lines: list[str] = []
            for gkey in sorted(groups):
                vals: list[float] = []
                for r in groups[gkey]:
                    raw = r.get(column)
                    if raw in (None, ""):
                        continue
                    try:
                        vals.append(float(raw))
                    except (TypeError, ValueError):
                        continue
                label = f"{gkey}: {column}" if group_by else column
                if not vals:
                    lines.append(f"{label} (no numeric values)")
                    continue
                if op == "count":
                    res: float | int = len(groups[gkey])
                elif op == "sum":
                    res = sum(vals)
                elif op == "avg":
                    res = sum(vals) / len(vals)
                elif op == "min":
                    res = min(vals)
                elif op == "max":
                    res = max(vals)
                else:
                    raise ValueError(f"unknown op: {op}")
                lines.append(f"{label} {op}={self._fmt(res)}")
            return "\n".join(lines) if lines else "no rows"
        raise ValueError(f"unknown action: {a}")

CurrentTimeTool

Bases: Tool

Get the current date and time.

Parameters:

Name	Type	Description	Default
config	dict | None	Optional dict with timezone — an IANA name such as "Europe/Moscow" (default "local").	None

Source code in teff/tool/builtin/fs.py

class CurrentTimeTool(Tool):
    """Get the current date and time.

    Args:
        config: Optional dict with ``timezone`` — an IANA name such as
            ``"Europe/Moscow"`` (default ``"local"``).
    """

    name = "current_time"
    description = "Get the current date and time"

    def __init__(self, config: dict | None = None):
        cfg = config or {}
        self.timezone = cfg.get("timezone", "local")

    def run(self, timezone: str = "") -> str:  # type: ignore[override]
        tz = timezone or self.timezone
        now = datetime.datetime.now(datetime.timezone.utc).astimezone()
        if tz != "local":
            try:
                now = now.astimezone(zoneinfo.ZoneInfo(tz))
            except zoneinfo.ZoneInfoNotFoundError as e:
                raise ValueError(f"unknown timezone: {tz}") from e
        return now.isoformat(timespec="seconds")

EditFileTool

Bases: Tool

Edit a file by replacing the first occurrence of a string.

Source code in teff/tool/builtin/file.py

class EditFileTool(Tool):
    """Edit a file by replacing the first occurrence of a string."""

    name = "edit_file"
    description = "Edit a file by replacing text"

    def run(self, path: str = "", old: str = "", new: str = "") -> str:  # type: ignore[override]
        with open(path) as f:
            content = f.read()
        if old not in content:
            raise ValueError(f"text not found in {path}")
        content = content.replace(old, new, 1)
        with open(path, "w") as f:
            f.write(content)
        return f"replaced in {path}"

GetEnvTool

Bases: Tool

Read an environment variable, masking secret-looking values.

Values whose names hint at credentials (TOKEN, API_KEY, PASSWORD, DSN, …) are returned as *** unless the tool is configured with mask_secrets=False.

Source code in teff/tool/builtin/fs.py

class GetEnvTool(Tool):
    """Read an environment variable, masking secret-looking values.

    Values whose names hint at credentials (``TOKEN``, ``API_KEY``,
    ``PASSWORD``, ``DSN``, …) are returned as ``***`` unless the tool is
    configured with ``mask_secrets=False``.
    """

    name = "getenv"
    description = "Read an environment variable (secrets are masked)"

    def __init__(self, config: dict | None = None):
        cfg = config or {}
        self.mask_secrets = cfg.get("mask_secrets", True)

    def run(self, name: str = "") -> str:  # type: ignore[override]
        if not name:
            raise ValueError("name is required")
        value = os.environ.get(name)
        if value is None:
            return "not set"
        if self.mask_secrets and self._is_secret(name):
            return "***"
        return value

    @staticmethod
    def _is_secret(name: str) -> bool:
        lowered = name.lower()
        return any(hint in lowered for hint in _SECRET_HINTS)

GitHubApproveTool

Bases: _GitHubBase

Approve a pull request by submitting an APPROVE review.

Parameters:

Name	Type	Description	Default
repo		owner/repo.	required
number		Pull request number.	required

Source code in teff/tool/builtin/github.py

class GitHubApproveTool(_GitHubBase):
    """Approve a pull request by submitting an APPROVE review.

    Args:
        repo: ``owner/repo``.
        number: Pull request number.
    """

    name = "github_approve"
    description = "Approve a GitHub pull request"

    async def arun(  # type: ignore[override]
        self, repo: str, number: str
    ) -> str:
        r = self._repo(repo)
        path = f"/{r}/pulls/{number}/reviews"
        await self._request(
            "POST", path, json_body={"event": "APPROVE", "body": "Approved"}
        )
        return f"approved PR #{number}"

GitHubGetPRChangesTool

Bases: _GitHubBase

Fetch a pull request's diff (changed files + per-file patches).

Parameters:

Name	Type	Description	Default
repo		owner/repo.	required
number		Pull request number (the #N).	required
max_chars		Cap on the returned diff text (default 20000).	required

Source code in teff/tool/builtin/github.py

class GitHubGetPRChangesTool(_GitHubBase):
    """Fetch a pull request's diff (changed files + per-file patches).

    Args:
        repo: ``owner/repo``.
        number: Pull request number (the ``#N``).
        max_chars: Cap on the returned diff text (default 20000).
    """

    name = "github_get_pr_changes"
    description = "Fetch the diff of a GitHub pull request"

    async def arun(  # type: ignore[override]
        self, repo: str, number: str, max_chars: int = 20000
    ) -> str:
        r = self._repo(repo)
        path = f"/{r}/pulls/{number}/files"
        text = await self._request("GET", path)
        data = json.loads(text)
        out = [f"# PR #{number}"]
        for change in data:
            out.append(
                f"\n== {change.get('filename', '')} "
                f"(+{change.get('additions', 0)} "
                f"-{change.get('deletions', 0)} {change.get('status', '')})"
            )
            patch = change.get("patch", "")
            out.append(patch[:max_chars])
        return "\n".join(out) if data else f"no changes for PR #{number}"

GitHubListOpenPRsTool

Bases: _GitHubBase

List open pull requests for an owner/repo.

Parameters:

Name	Type	Description	Default
repo		owner/repo.	required
limit		Maximum number of PRs to return (default 50).	required
state		PR state filter (default open).	required

Source code in teff/tool/builtin/github.py

class GitHubListOpenPRsTool(_GitHubBase):
    """List open pull requests for an ``owner/repo``.

    Args:
        repo: ``owner/repo``.
        limit: Maximum number of PRs to return (default 50).
        state: PR state filter (default ``open``).
    """

    name = "github_list_open_prs"
    description = "List open pull requests for a GitHub repository"

    async def arun(  # type: ignore[override]
        self, repo: str, limit: int = 50, state: str = "open"
    ) -> str:
        r = self._repo(repo)
        path = f"/{r}/pulls?state={state}&per_page={limit}"
        text = await self._request("GET", path)
        data = json.loads(text)
        lines = []
        for pr in data:
            lines.append(
                f"#{pr['number']}\t{pr.get('state', '')}\t"
                f"{pr.get('title', '')}\t(pr_id={pr.get('id')})"
            )
        return "\n".join(lines) if lines else "no open pull requests"

GitHubPostCommentTool

Bases: _GitHubBase

Post a comment on a pull request (as an issue comment).

Parameters:

Name	Type	Description	Default
repo		owner/repo.	required
number		Pull request number.	required
body		The comment text to post.	required

Source code in teff/tool/builtin/github.py

class GitHubPostCommentTool(_GitHubBase):
    """Post a comment on a pull request (as an issue comment).

    Args:
        repo: ``owner/repo``.
        number: Pull request number.
        body: The comment text to post.
    """

    name = "github_post_comment"
    description = "Post a comment on a GitHub pull request"

    async def arun(  # type: ignore[override]
        self, repo: str, number: str, body: str
    ) -> str:
        r = self._repo(repo)
        path = f"/{r}/issues/{number}/comments"
        text = await self._request("POST", path, json_body={"body": body})
        data = json.loads(text)
        return f"comment posted on #{number} (comment_id={data.get('id')})"

GitLabApproveTool

Bases: _GitLabBase

Approve a merge request.

Parameters:

Name	Type	Description	Default
project		Project id or URL-encoded path.	required
iid		Merge request internal id.	required

Source code in teff/tool/builtin/gitlab.py

class GitLabApproveTool(_GitLabBase):
    """Approve a merge request.

    Args:
        project: Project id or URL-encoded path.
        iid: Merge request internal id.
    """

    name = "gitlab_approve"
    description = "Approve a GitLab merge request"

    async def arun(  # type: ignore[override]
        self, project: str, iid: str
    ) -> str:
        pid = self._project_id(project)
        path = f"/projects/{pid}/merge_requests/{iid}/approve"
        import httpx

        self._require()
        url = f"{self.url}/api/v4{path}"
        headers = {"PRIVATE-TOKEN": self.token}
        async with httpx.AsyncClient(timeout=30) as client:
            response = await client.post(url, headers=headers)
            if response.status_code >= 400:
                raise ValueError(
                    f"GitLab POST {path} -> HTTP {response.status_code}: "
                    f"{response.text[:500]}"
                )
        return f"approved MR !{iid}"

GitLabGetMRChangesTool

Bases: _GitLabBase

Fetch a merge request's diff (changed files + line-level changes).

Parameters:

Name	Type	Description	Default
project		Project id or URL-encoded path.	required
iid		Merge request internal id (the !N number).	required
max_chars		Cap on the returned diff text (default 20000).	required

Source code in teff/tool/builtin/gitlab.py

class GitLabGetMRChangesTool(_GitLabBase):
    """Fetch a merge request's diff (changed files + line-level changes).

    Args:
        project: Project id or URL-encoded path.
        iid: Merge request internal id (the ``!N`` number).
        max_chars: Cap on the returned diff text (default 20000).
    """

    name = "gitlab_get_mr_changes"
    description = "Fetch the diff of a GitLab merge request"

    async def arun(  # type: ignore[override]
        self, project: str, iid: str, max_chars: int = 20000
    ) -> str:
        pid = self._project_id(project)
        path = f"/projects/{pid}/merge_requests/{iid}/changes"
        text = await self._request("GET", path)
        data = json.loads(text)
        changes = data.get("changes", [])
        out = [f"# MR !{iid}: {data.get('title', '')}"]
        out.append(
            f"state={data.get('state', '')}  target_branch={data.get('target_branch', '')}"
        )
        for change in changes:
            out.append(
                f"\n== {change.get('new_path', '')} "
                f"(+{change.get('new_file', False)} "
                f"-{change.get('deleted_file', False)})"
            )
            diff = change.get("diff", "")
            out.append(diff[:max_chars])
        return "\n".join(out)

GitLabListOpenMRsTool

Bases: _GitLabBase

List open merge requests for a project.

Parameters:

Name	Type	Description	Default
project		Project id or URL-encoded path (group/repo).	required
limit		Maximum number of MRs to return (default 50).	required
state		MR state filter (default opened).	required

Source code in teff/tool/builtin/gitlab.py

class GitLabListOpenMRsTool(_GitLabBase):
    """List open merge requests for a project.

    Args:
        project: Project id or URL-encoded path (``group/repo``).
        limit: Maximum number of MRs to return (default 50).
        state: MR state filter (default ``opened``).
    """

    name = "gitlab_list_open_mrs"
    description = "List open merge requests for a GitLab project"

    async def arun(  # type: ignore[override]
        self, project: str, limit: int = 50, state: str = "opened"
    ) -> str:
        pid = self._project_id(project)
        path = f"/projects/{pid}/merge_requests?state={state}&per_page={limit}"
        text = await self._request("GET", path)
        data = json.loads(text)
        lines = []
        for mr in data:
            lines.append(
                f"!{mr['iid']}\t{mr.get('state', '')}\t"
                f"{mr.get('title', '')}\t(mr_id={mr.get('id')})"
            )
        return "\n".join(lines) if lines else "no open merge requests"

GitLabPostNoteTool

Bases: _GitLabBase

Post a note (comment) on a merge request.

Parameters:

Name	Type	Description	Default
project		Project id or URL-encoded path.	required
iid		Merge request internal id.	required
body		The note text to post.	required

Source code in teff/tool/builtin/gitlab.py

class GitLabPostNoteTool(_GitLabBase):
    """Post a note (comment) on a merge request.

    Args:
        project: Project id or URL-encoded path.
        iid: Merge request internal id.
        body: The note text to post.
    """

    name = "gitlab_post_note"
    description = "Post a comment/note on a GitLab merge request"

    async def arun(  # type: ignore[override]
        self, project: str, iid: str, body: str
    ) -> str:
        pid = self._project_id(project)
        path = f"/projects/{pid}/merge_requests/{iid}/notes"
        import httpx

        self._require()
        url = f"{self.url}/api/v4{path}"
        headers = {"PRIVATE-TOKEN": self.token}
        async with httpx.AsyncClient(timeout=30) as client:
            response = await client.post(url, headers=headers, json={"body": body})
            if response.status_code >= 400:
                raise ValueError(
                    f"GitLab POST {path} -> HTTP {response.status_code}: "
                    f"{response.text[:500]}"
                )
            data = response.json()
        return f"note posted on !{iid} (note_id={data.get('id')})"

GitTool

Bases: Tool

Inspect a git repository without mutating it.

Parameters:

Name	Type	Description	Default
action		status | log | diff | ls_files | branch | show.	required
limit		Max commits for log (default 20).	required
ref		Commit/ref for diff/show (default HEAD for show; empty for diff means working tree).	required
path		Restrict diff/ls_files to a path.	required
max_chars		Cap on returned output (default 20000).	required

Args (config): path: Repository directory (default .).

Source code in teff/tool/builtin/git.py

class GitTool(Tool):
    """Inspect a git repository without mutating it.

    Args:
        action: ``status`` | ``log`` | ``diff`` | ``ls_files``
            | ``branch`` | ``show``.
        limit: Max commits for ``log`` (default 20).
        ref: Commit/ref for ``diff``/``show`` (default ``HEAD`` for
            ``show``; empty for ``diff`` means working tree).
        path: Restrict ``diff``/``ls_files`` to a path.
        max_chars: Cap on returned output (default 20000).

    Args (config):
        path: Repository directory (default ``.``).
    """

    name = "git"
    description = (
        "Inspect a git repository read-only (status, log, diff, ls_files, branch, show)"
    )

    _READ_ONLY_ACTIONS = ("status", "log", "diff", "ls_files", "branch", "show")

    def __init__(self, config: dict | None = None):
        cfg = config or {}
        self.path = cfg.get("path", ".")

    def _git(self, *args: str) -> str:
        cmd = ["git", "-C", self.path, *args]
        proc = subprocess.run(cmd, capture_output=True, text=True)
        if proc.returncode != 0:
            stderr = (proc.stderr or "").strip()
            raise ValueError(f"git {' '.join(args)} failed: {stderr[:500]}")
        return proc.stdout

    def run(  # type: ignore[override]
        self,
        action: str,
        limit: int = 20,
        ref: str = "",
        path: str = "",
        max_chars: int = 20000,
    ) -> str:
        if not action:
            raise ValueError("action is required (status, log, diff, ...)")
        a = action.lower()
        if a not in self._READ_ONLY_ACTIONS:
            raise ValueError(f"unknown action: {action}")
        if a == "status":
            out = self._git("status", "--short")
            return out.strip()[:max_chars] or "clean working tree"
        if a == "log":
            out = self._git("log", "-n", str(int(limit)), "--oneline", "--decorate")
            return out.strip()[:max_chars] or "no commits"
        if a == "diff":
            args = ["diff"]
            if ref:
                args.append(ref)
            if path:
                args.extend(["--", path])
            out = self._git(*args)
            return out[:max_chars] or "no changes"
        if a == "ls_files":
            args = ["ls-files"]
            if path:
                args.append(path)
            out = self._git(*args)
            return out.strip()[:max_chars] or "no files"
        if a == "branch":
            out = self._git("branch", "-a")
            return out.strip()[:max_chars] or "no branches"
        if a == "show":
            out = self._git("show", ref or "HEAD")
            return out[:max_chars] or "nothing to show"
        raise ValueError(f"unknown action: {action}")

GlobTool

Bases: Tool

Find files matching a glob pattern.

Source code in teff/tool/builtin/fs.py

class GlobTool(Tool):
    """Find files matching a glob pattern."""

    name = "glob"
    description = "Find files matching a glob pattern"

    def run(self, pattern: str = "") -> str:  # type: ignore[override]
        if not pattern:
            raise ValueError("pattern is required")
        matches = sorted(glob.glob(pattern, recursive=True))
        return "\n".join(matches) if matches else "no matches"

HttpRequestTool

Bases: Tool

Send an HTTP request to an API endpoint.

Unlike :class:~teff.tool.builtin.web_fetch.WebFetchTool, this tool exposes the full request surface: method, headers, and body, and returns the raw response (status, headers, text).

Parameters:

Name	Type	Description	Default
config	dict | None	Optional dict with timeout (seconds, default 30).	None

Source code in teff/tool/builtin/http.py

class HttpRequestTool(Tool):
    """Send an HTTP request to an API endpoint.

    Unlike :class:`~teff.tool.builtin.web_fetch.WebFetchTool`, this tool
    exposes the full request surface: method, headers, and body, and
    returns the raw response (status, headers, text).

    Args:
        config: Optional dict with ``timeout`` (seconds, default 30).
    """

    name = "http_request"
    description = "Send an HTTP request (GET/POST/PUT/DELETE) and return the response"

    def __init__(self, config: dict | None = None):
        cfg = config or {}
        self.timeout = cfg.get("timeout", 30.0)

    async def arun(  # type: ignore[override]
        self,
        url: str = "",
        method: str = "GET",
        headers: str = "",
        body: str = "",
        max_chars: int = 20000,
    ) -> str:
        if not url:
            raise ValueError("url is required")
        import httpx

        request_headers: dict | None = None
        if headers:
            try:
                request_headers = json.loads(headers)
            except json.JSONDecodeError as e:
                raise ValueError("headers must be a JSON object string") from e

        async with httpx.AsyncClient(timeout=self.timeout) as client:
            response = await client.request(
                method.upper(),
                url,
                headers=request_headers,
                content=body.encode("utf-8") if body else None,
            )

        response_headers = "\n".join(f"{k}: {v}" for k, v in response.headers.items())
        text = response.text[:max_chars]
        return f"HTTP {response.status_code}\n{response_headers}\n\n{text}"

JsonParseTool

Bases: Tool

Parse a JSON string and pretty-print it (or validate it).

Source code in teff/tool/builtin/data.py

class JsonParseTool(Tool):
    """Parse a JSON string and pretty-print it (or validate it)."""

    name = "json_parse"
    description = "Parse and pretty-print a JSON string"

    def run(self, text: str = "", indent: int = 2) -> str:  # type: ignore[override]
        if not text:
            raise ValueError("text is required")
        try:
            data = json.loads(text)
        except json.JSONDecodeError as e:
            raise ValueError(f"invalid JSON: {e}") from e
        return json.dumps(data, ensure_ascii=False, indent=indent)

KVStoreTool

Bases: Tool

A persistent JSON-backed key-value store.

Data lives in a single JSON file (config key path). Operations are selected with action: get, set, delete, list.

Parameters:

Name	Type	Description	Default
config	dict | None	Optional dict with path (default ./kv_store.json).	None

Source code in teff/tool/builtin/data.py

class KVStoreTool(Tool):
    """A persistent JSON-backed key-value store.

    Data lives in a single JSON file (config key ``path``). Operations
    are selected with ``action``: ``get``, ``set``, ``delete``, ``list``.

    Args:
        config: Optional dict with ``path`` (default ``./kv_store.json``).
    """

    name = "kv_store"
    description = "Read/write a persistent key-value store (get, set, delete, list)"

    def __init__(self, config: dict | None = None):
        cfg = config or {}
        self.path = cfg.get("path", "./kv_store.json")
        self._data: dict = {}
        self._load()

    def _load(self) -> None:
        if os.path.exists(self.path):
            with open(self.path, encoding="utf-8") as f:
                self._data = json.load(f)
        else:
            self._data = {}

    def _save(self) -> None:
        with open(self.path, "w", encoding="utf-8") as f:
            json.dump(self._data, f, ensure_ascii=False, indent=2)

    def run(  # type: ignore[override]
        self, action: str = "get", key: str = "", value: str = ""
    ) -> str:
        if action == "get":
            if key not in self._data:
                return "not found"
            return json.dumps(self._data[key], ensure_ascii=False)
        if action == "set":
            if not key:
                raise ValueError("key is required")
            try:
                parsed = json.loads(value)
            except json.JSONDecodeError:
                parsed = value
            self._data[key] = parsed
            self._save()
            return f"set {key}"
        if action == "delete":
            if key in self._data:
                del self._data[key]
                self._save()
                return f"deleted {key}"
            return "not found"
        if action == "list":
            return "\n".join(sorted(self._data.keys())) if self._data else "empty"
        raise ValueError(f"unknown action: {action}")

ListDirTool

Bases: Tool

List the contents of a directory.

Source code in teff/tool/builtin/fs.py

class ListDirTool(Tool):
    """List the contents of a directory."""

    name = "list_dir"
    description = "List files and directories in a path"

    def run(self, path: str = ".", recursive: bool = False) -> str:  # type: ignore[override]
        if not os.path.isdir(path):
            raise ValueError(f"not a directory: {path}")
        entries: list[str] = []
        if recursive:
            for root, dirs, files in os.walk(path):
                for name in sorted(dirs + files):
                    entries.append(os.path.join(root, name))
        else:
            entries = sorted(os.listdir(path))
        return "\n".join(entries) if entries else "empty directory"

LockTool

Bases: _RedisBase

Distributed lock over Redis (KeyDB/Valkey supported).

Parameters:

Name	Type	Description	Default
action		acquire | release | renew | status.	required
key		Lock name.	required
ttl		Lease length in seconds for acquire (default 30) or the new lease for renew.	required

Args (config): same as the redis tool — url or host/port/db/password/username.

Source code in teff/tool/builtin/lock.py

class LockTool(_RedisBase):
    """Distributed lock over Redis (KeyDB/Valkey supported).

    Args:
        action: ``acquire`` | ``release`` | ``renew`` | ``status``.
        key: Lock name.
        ttl: Lease length in seconds for ``acquire`` (default 30) or the
            new lease for ``renew``.

    Args (config): same as the ``redis`` tool — ``url`` or
        ``host``/``port``/``db``/``password``/``username``.
    """

    name = "lock"
    description = (
        "Distributed lock over Redis-compatible stores (acquire, release, "
        "renew, status)"
    )

    def __init__(self, config: dict | None = None):
        super().__init__(config)
        self.token = uuid.uuid4().hex

    def run(  # type: ignore[override]
        self, action: str, key: str = "", ttl: int = 30
    ) -> str:
        if not action:
            raise ValueError("action is required (acquire, release, renew, status)")
        if not key:
            raise ValueError("key is required")
        client = self._client()
        try:
            a = action.lower()
            if a == "acquire":
                if int(ttl) <= 0:
                    raise ValueError("ttl must be > 0")
                ok = client.set(key, self.token, nx=True, ex=int(ttl))
                return "acquired" if ok else "held by someone else"
            if a == "release":
                released = client.eval(_DEL_IF_MATCH, 1, key, self.token)
                return "released" if released else "not held (or owned by someone else)"
            if a == "renew":
                renewed = client.eval(_EXPIRE_IF_MATCH, 1, key, self.token, int(ttl))
                return (
                    f"renewed {key} for {ttl}s"
                    if renewed
                    else "not held (or owned by someone else)"
                )
            if a == "status":
                holder = client.get(key)
                if holder is None:
                    return f"{key} is free"
                remaining = client.ttl(key)
                who = (
                    "me"
                    if holder == self.token
                    else f"another holder ({holder[:8]}...)"
                )
                return f"{key} held by {who} ({remaining}s left)"
            raise ValueError(f"unknown action: {a}")
        finally:
            client.close()

MemoryTool

Bases: Tool

Tool that lets an agent read and write long-term memory.

Usage::

memory = MemoryTool(
    store=SQLiteVectorStore(path="./memory.db", dim=768),
    embedder=Embedder(provider="ollama", model="nomic-embed-text"),
    namespace=("users", "u1"),
)
await memory.arun(action="remember", text="prefers email over Slack")
result = await memory.arun(action="recall", query="how to reach them?")

Actions (passed as action):

• remember — upsert a fact (text plus optional metadata). When similarity_threshold is set and a semantically close item already exists in the namespace, the new text overwrites that item instead of creating a duplicate.
• recall — return top-k memories for a query (or the most recent if no query is given), formatted for a prompt.
• forget — delete the memory at key.
• list — enumerate stored keys.

Can be built from a config dict (e.g. a tools: entry in a workflow YAML)::

{
  "name": "memory",
  "store": {"type": "sqlite", "path": "./memory.db", "dim": 768},
  "embedder": {"provider": "ollama", "model": "nomic-embed-text"},
  "namespace": ["users", "${USER_ID}"],
  "default_k": 5,
  "similarity_threshold": 0.6,
}

Supported store types match RAGTool: in_memory (default), sqlite, chroma, qdrant, pgvector, faiss, lance, milvus, weaviate, pinecone.

Methods:

Name	Description
arun	Run a memory operation and return a human-readable result.

Source code in teff/memory/tool.py

class MemoryTool(Tool):
    """Tool that lets an agent read and write long-term memory.

    Usage::

        memory = MemoryTool(
            store=SQLiteVectorStore(path="./memory.db", dim=768),
            embedder=Embedder(provider="ollama", model="nomic-embed-text"),
            namespace=("users", "u1"),
        )
        await memory.arun(action="remember", text="prefers email over Slack")
        result = await memory.arun(action="recall", query="how to reach them?")

    Actions (passed as ``action``):

    - ``remember`` — upsert a fact (``text`` plus optional ``metadata``).
      When ``similarity_threshold`` is set and a semantically close item
      already exists in the namespace, the new text overwrites that item
      instead of creating a duplicate.
    - ``recall`` — return top-*k* memories for a ``query`` (or the most
      recent if no query is given), formatted for a prompt.
    - ``forget`` — delete the memory at ``key``.
    - ``list`` — enumerate stored keys.

    Can be built from a config dict (e.g. a ``tools:`` entry in a
    workflow YAML)::

        {
          "name": "memory",
          "store": {"type": "sqlite", "path": "./memory.db", "dim": 768},
          "embedder": {"provider": "ollama", "model": "nomic-embed-text"},
          "namespace": ["users", "${USER_ID}"],
          "default_k": 5,
          "similarity_threshold": 0.6,
        }

    Supported store types match ``RAGTool``: ``in_memory`` (default),
    ``sqlite``, ``chroma``, ``qdrant``, ``pgvector``, ``faiss``, ``lance``,
    ``milvus``, ``weaviate``, ``pinecone``.
    """

    name = "memory"
    description = (
        "Long-term memory: remember facts, recall relevant memories, "
        "forget, and list what is stored."
    )

    def __init__(
        self,
        config: dict | None = None,
        *,
        store: VectorStore | None = None,
        embedder: Embedder | None = None,
        namespace: tuple[str, ...] | list[str] = (),
        default_k: int = 5,
        similarity_threshold: float | None = None,
        ttl: float | None = None,
    ):
        self._memory: MemoryStore | None = None
        self._namespace = tuple(namespace)
        self._default_k = default_k
        self._threshold = similarity_threshold
        self._ttl = ttl
        if isinstance(config, dict):
            self._apply_config(config)
        elif store is not None and embedder is not None:
            self.memory = MemoryStore(store=store, embedder=embedder, ttl=ttl)

    @property
    def memory(self) -> MemoryStore:
        if self._memory is None:
            raise RuntimeError("memory store not initialised")
        return self._memory

    @memory.setter
    def memory(self, value: MemoryStore) -> None:
        self._memory = value

    def _apply_config(self, config: dict) -> None:
        self.memory = memory_from_config(config, default_ttl=self._ttl)
        ns = config.get("namespace")
        if ns:
            self._namespace = tuple(str(part) for part in ns)
        if config.get("default_k") is not None:
            self._default_k = int(config["default_k"])
        if config.get("similarity_threshold") is not None:
            self._threshold = float(config["similarity_threshold"])

    async def arun(  # type: ignore[override]
        self,
        action: str = "recall",
        key: str = "",
        text: str = "",
        value: dict | None = None,
        query: str = "",
        metadata: dict | None = None,
        k: int | None = None,
    ) -> str:
        """Run a memory operation and return a human-readable result.

        The namespace is fixed at construction time and can never be
        overridden by the caller — an agent cannot address another owner's
        memories by passing a namespace.  Per-owner isolation is achieved by
        building one tool per owner (``namespace=("users", owner)``).
        """
        ns = self._namespace
        eff_k = int(k) if k is not None else self._default_k
        mem = self.memory

        if action == "remember":
            return await self._remember(ns, key, text, value, metadata)
        if action == "recall":
            items = await mem.search(ns, query=query or None, k=eff_k)
            return _format_recall(items)
        if action == "forget":
            if not key:
                return "forget requires a `key`"
            await mem.delete(ns, key)
            return f"forgotten {key!r}"
        if action == "list":
            keys = await mem.list(ns, limit=1000)
            return "\n".join(keys) if keys else "(no memories)"
        raise ValueError(f"unknown memory action: {action!r}")

    async def _remember(
        self,
        ns: tuple[str, ...],
        key: str,
        text: str,
        value: dict | None,
        metadata: dict | None,
    ) -> str:
        if value is None:
            if not text:
                return "remember requires `text`"
            value = {"text": text, **(metadata or {})}
        elif "text" not in value:
            return "remember `value` requires a 'text' field"

        if self._threshold is not None:
            similar = await self.memory.search(ns, query=value["text"], k=1)
            if (
                similar
                and similar[0].score is not None
                and similar[0].score >= self._threshold
            ):
                key = similar[0].key

        final_key = key or uuid.uuid4().hex[:12]
        await self.memory.put(ns, final_key, value)
        return f"remembered {final_key!r}"

arun async

arun(action='recall', key='', text='', value=None, query='', metadata=None, k=None)

Run a memory operation and return a human-readable result.

The namespace is fixed at construction time and can never be overridden by the caller — an agent cannot address another owner's memories by passing a namespace. Per-owner isolation is achieved by building one tool per owner (namespace=("users", owner)).

Source code in teff/memory/tool.py

async def arun(  # type: ignore[override]
    self,
    action: str = "recall",
    key: str = "",
    text: str = "",
    value: dict | None = None,
    query: str = "",
    metadata: dict | None = None,
    k: int | None = None,
) -> str:
    """Run a memory operation and return a human-readable result.

    The namespace is fixed at construction time and can never be
    overridden by the caller — an agent cannot address another owner's
    memories by passing a namespace.  Per-owner isolation is achieved by
    building one tool per owner (``namespace=("users", owner)``).
    """
    ns = self._namespace
    eff_k = int(k) if k is not None else self._default_k
    mem = self.memory

    if action == "remember":
        return await self._remember(ns, key, text, value, metadata)
    if action == "recall":
        items = await mem.search(ns, query=query or None, k=eff_k)
        return _format_recall(items)
    if action == "forget":
        if not key:
            return "forget requires a `key`"
        await mem.delete(ns, key)
        return f"forgotten {key!r}"
    if action == "list":
        keys = await mem.list(ns, limit=1000)
        return "\n".join(keys) if keys else "(no memories)"
    raise ValueError(f"unknown memory action: {action!r}")

PDFReadTool

Bases: Tool

Extract text from a PDF file.

Requires pypdf (from teff[tools]). Returns the text of each page, optionally limited to max_chars characters.

Parameters:

Name	Type	Description	Default
config	dict | None	Optional dict. Currently unused, kept for config parity.	None

Source code in teff/tool/builtin/pdf.py

class PDFReadTool(Tool):
    """Extract text from a PDF file.

    Requires ``pypdf`` (from ``teff[tools]``). Returns the text of each
    page, optionally limited to *max_chars* characters.

    Args:
        config: Optional dict. Currently unused, kept for config parity.
    """

    name = "read_pdf"
    description = "Extract text from a PDF file"

    def __init__(self, config: dict | None = None):
        pass

    def run(self, path: str = "", max_chars: int = 50000) -> str:  # type: ignore[override]
        if not path:
            raise ValueError("path is required")
        try:
            from pypdf import PdfReader
        except ImportError as e:
            msg = "read_pdf requires 'pypdf' (pip install teff[tools])"
            raise ImportError(msg) from e

        reader = PdfReader(path)
        parts: list[str] = []
        for i, page in enumerate(reader.pages, 1):
            text = page.extract_text() or ""
            parts.append(f"--- page {i} ---\n{text}")
        result = "\n".join(parts).strip()
        if not result:
            return "no text found in pdf"
        return result[:max_chars]

PythonEvalTool

Bases: Tool

Safely evaluate a Python expression using an AST whitelist.

Supports numbers, arithmetic, math constants/functions, strings, lists, tuples, dicts, and comparisons. Imports, attributes beyond math., and calls outside the whitelist are rejected.

Source code in teff/tool/builtin/data.py

class PythonEvalTool(Tool):
    """Safely evaluate a Python expression using an AST whitelist.

    Supports numbers, arithmetic, ``math`` constants/functions, strings,
    lists, tuples, dicts, and comparisons. Imports, attributes beyond
    ``math.``, and calls outside the whitelist are rejected.
    """

    name = "python_eval"
    description = "Safely evaluate a Python expression"

    _BUILTIN_FUNCS: dict[str, Callable[..., Any]] = {
        "abs": abs,
        "len": len,
        "min": min,
        "max": max,
        "sum": sum,
        "round": round,
        "range": range,
        "int": int,
        "float": float,
        "str": str,
        "bool": bool,
        "list": list,
        "tuple": tuple,
        "dict": dict,
        "set": set,
        "sorted": sorted,
        "enumerate": enumerate,
        "zip": zip,
        "isinstance": isinstance,
    }

    def run(self, expression: str = "") -> str:  # type: ignore[override]
        if not expression:
            raise ValueError("expression is required")
        tree = ast.parse(expression, mode="eval")
        return str(self._eval(tree.body))

    def _eval(self, node) -> object:
        import math

        if isinstance(node, ast.Constant):
            return node.value
        if isinstance(node, ast.BinOp):
            binop = _ALLOWED_BINOPS.get(type(node.op))
            if binop is None:
                raise ValueError(f"operator not allowed: {type(node.op).__name__}")
            return binop(self._eval(node.left), self._eval(node.right))
        if isinstance(node, ast.UnaryOp):
            unop = _ALLOWED_UNARYOPS.get(type(node.op))
            if unop is None:
                raise ValueError(f"operator not allowed: {type(node.op).__name__}")
            return unop(self._eval(node.operand))
        if isinstance(node, ast.Name):
            if node.id in _MATH_CONSTANTS:
                return getattr(math, _MATH_CONSTANTS[node.id])
            if node.id in ("True", "False", "None"):
                return {"True": True, "False": False, "None": None}[node.id]
            raise ValueError(f"name not allowed: {node.id}")
        if isinstance(node, ast.Attribute):
            if isinstance(node.value, ast.Name) and node.value.id == "math":
                return getattr(math, node.attr)
            raise ValueError(f"attribute not allowed: {node.attr}")
        if isinstance(node, ast.Call):
            if isinstance(node.func, ast.Name):
                fn = self._BUILTIN_FUNCS.get(node.func.id)
                if fn is None:
                    raise ValueError(f"function not allowed: {node.func.id}")
                return fn(*(self._eval(a) for a in node.args))
            if isinstance(node.func, ast.Attribute) and isinstance(
                node.func.value, ast.Name
            ):
                if node.func.value.id == "math":
                    fn = getattr(math, node.func.attr, None)
                    if fn is None:
                        raise ValueError(f"function not allowed: {node.func.attr}")
                    return fn(*(self._eval(a) for a in node.args))
            raise ValueError("call not allowed")
        if isinstance(node, ast.List):
            return [self._eval(e) for e in node.elts]
        if isinstance(node, ast.Tuple):
            return tuple(self._eval(e) for e in node.elts)
        if isinstance(node, ast.Dict):
            return {
                self._eval(k): self._eval(v)
                for k, v in zip(node.keys, node.values)
                if k is not None
            }
        if isinstance(node, ast.Compare):
            left: Any = self._eval(node.left)
            for cmp, comparator in zip(node.ops, node.comparators):
                right: Any = self._eval(comparator)
                if isinstance(cmp, ast.Eq):
                    ok = left == right
                elif isinstance(cmp, ast.NotEq):
                    ok = left != right
                elif isinstance(cmp, ast.Lt):
                    ok = left < right
                elif isinstance(cmp, ast.LtE):
                    ok = left <= right
                elif isinstance(cmp, ast.Gt):
                    ok = left > right
                elif isinstance(cmp, ast.GtE):
                    ok = left >= right
                else:
                    raise ValueError(f"operator not allowed: {type(cmp).__name__}")
                if not ok:
                    return False
                left = right
            return True
        if isinstance(node, ast.BoolOp):
            if isinstance(node.op, ast.And):
                result = True
                for v in node.values:
                    result = result and bool(self._eval(v))
            elif isinstance(node.op, ast.Or):
                result = False
                for v in node.values:
                    result = result or bool(self._eval(v))
            else:
                raise ValueError("boolean operator not allowed")
            return result
        raise ValueError(f"expression not allowed: {ast.dump(node)}")

RAGIngestTool

Bases: Tool

Add documents to a vector store.

Parameters:

Name	Type	Description	Default
text		Raw document text to chunk, embed and store (inline content).	required
path		File to load instead of text (see config type).	required
source_id		Optional stable id for the document (default: derived).	required
metadata		Extra metadata dict merged into every chunk.	required

Args (config): embedder: Embedder config (same shape as the rag tool). store: Vector-store config (same shape as the rag tool). chunker: Optional chunker kwargs. type: Loader for path — csv (default), txt, pdf, excel. Ignored when text is provided. text_column: Column used as text when loading a CSV/Excel file. parent_chunks: Keep full parent text per chunk (default false).

At least one of text or path must be supplied per call. The result is a short confirmation with the number of chunks written.

Source code in teff/tool/builtin/rag_ingest.py

class RAGIngestTool(Tool):
    """Add documents to a vector store.

    Args:
        text: Raw document text to chunk, embed and store (inline content).
        path: File to load instead of *text* (see config ``type``).
        source_id: Optional stable id for the document (default: derived).
        metadata: Extra metadata dict merged into every chunk.

    Args (config):
        embedder: Embedder config (same shape as the ``rag`` tool).
        store: Vector-store config (same shape as the ``rag`` tool).
        chunker: Optional chunker kwargs.
        type: Loader for ``path`` — ``csv`` (default), ``txt``, ``pdf``,
            ``excel``.  Ignored when ``text`` is provided.
        text_column: Column used as text when loading a CSV/Excel file.
        parent_chunks: Keep full parent text per chunk (default false).

    At least one of ``text`` or ``path`` must be supplied per call.  The
    result is a short confirmation with the number of chunks written.
    """

    name = "rag_ingest"
    description = (
        "Add a document to the knowledge base: give 'text' (content) or "
        "'path' (a csv/txt/pdf/excel file); it is chunked, embedded and "
        "stored for later 'rag' searches."
    )

    def __init__(self, config: dict | None = None):
        cfg = config or {}
        self.embedder = embedder_from_config(cfg)

        from teff.rag.stores.factory import store_from_config

        self.store = store_from_config(cfg.get("store") or {})
        self.chunker = Chunker(**(cfg.get("chunker") or {}))
        self.loader_type = cfg.get("type", "csv")
        self.text_column = cfg.get("text_column", "text")
        self.parent_chunks = bool(cfg.get("parent_chunks", False))

    async def arun(  # type: ignore[override]
        self,
        text: str = "",
        path: str = "",
        source_id: str | None = None,
        metadata: dict | None = None,
    ) -> str:
        docs: list[tuple[str, dict]] = []
        if text.strip():
            docs.append((text, {"id": source_id or "inline"}))
        elif path:
            loader = _DOCUMENT_LOADERS.get(self.loader_type)
            if loader is None:
                msg = f"unsupported document type: {self.loader_type}"
                raise ValueError(msg)
            kwargs: dict = {"path": path}
            if self.loader_type in ("csv", "excel"):
                kwargs["text_column"] = self.text_column
            docs = loader(**kwargs)
            if not docs:
                return f"no documents loaded from {path}"
        else:
            raise ValueError("rag_ingest requires 'text' or 'path'")

        extra = metadata or {}
        total = 0
        for doc_text, doc_meta in docs:
            meta = {**doc_meta, **extra}
            if source_id:
                meta["id"] = source_id
            await self._store(doc_text, meta)
            total += 1

        return f"ingested {total} document(s), {len(docs)} chunked+embedded"

    async def _store(self, text: str, metadata: dict) -> None:
        import uuid

        chunks = self.chunker.chunk(text)
        parent_id = metadata.get("id") or f"doc_{uuid.uuid4().hex[:8]}"
        embeddings = await self.embedder.embed_many(chunks)
        vectors = []
        for i, (chunk, vec) in enumerate(zip(chunks, embeddings)):
            if self.parent_chunks:
                doc_id = f"{parent_id}_{i}"
                meta = {
                    **metadata,
                    "id": parent_id,
                    "parent_id": parent_id,
                    "parent_text": text,
                    "text": chunk,
                    "chunk_index": i,
                }
            else:
                doc_id = f"{parent_id}_{i}"
                meta = {**metadata, "text": chunk, "chunk_index": i}
            vectors.append((doc_id, vec, meta))
        await self.store.add(vectors)

ReadFileTool

Bases: Tool

Read a file's contents.

Source code in teff/tool/builtin/file.py

class ReadFileTool(Tool):
    """Read a file's contents."""

    name = "read_file"
    description = "Read a file"

    def run(self, path: str = "") -> str:  # type: ignore[override]
        with open(path) as f:
            return f.read()

RedisTool

Bases: _RedisBase

Read/write a Redis-compatible store (KeyDB/Valkey supported).

Parameters:

Name	Type	Description	Default
action		ping | get | set | delete | exists | list | ttl | expire | incr | rpush | lrange | sadd | smembers | hset | hget | hgetall | publish.	required
key		Key to operate on (not needed for ping/list).	required
value		Value for set/rpush/hset.	required
ttl		Expiry in seconds for set (ex) or expire.	required
pattern		Glob pattern for list (default *).	required
field		Hash field for hset/hget.	required
member		Set member for sadd.	required
channel		Channel for publish.	required
message		Message for publish.	required
start, stop		Range for lrange (default all).	required

Source code in teff/tool/builtin/redis.py

class RedisTool(_RedisBase):
    """Read/write a Redis-compatible store (KeyDB/Valkey supported).

    Args:
        action: ``ping`` | ``get`` | ``set`` | ``delete`` | ``exists``
            | ``list`` | ``ttl`` | ``expire`` | ``incr`` | ``rpush``
            | ``lrange`` | ``sadd`` | ``smembers`` | ``hset`` | ``hget``
            | ``hgetall`` | ``publish``.
        key: Key to operate on (not needed for ``ping``/``list``).
        value: Value for ``set``/``rpush``/``hset``.
        ttl: Expiry in seconds for ``set`` (``ex``) or ``expire``.
        pattern: Glob pattern for ``list`` (default ``*``).
        field: Hash field for ``hset``/``hget``.
        member: Set member for ``sadd``.
        channel: Channel for ``publish``.
        message: Message for ``publish``.
        start, stop: Range for ``lrange`` (default all).
    """

    name = "redis"
    description = (
        "Read/write a Redis-compatible key-value store (get, set, delete, "
        "list, exists, ttl, expire, incr, lists, sets, hashes, publish)"
    )

    def run(  # type: ignore[override]
        self,
        action: str,
        key: str = "",
        value: str = "",
        ttl: int = -1,
        pattern: str = "*",
        field: str = "",
        member: str = "",
        channel: str = "",
        message: str = "",
        start: int = 0,
        stop: int = -1,
    ) -> str:
        if not action:
            raise ValueError("action is required (get, set, delete, list, ...)")
        client = self._client()
        try:
            a = action.lower()
            if a == "ping":
                return "PONG" if client.ping() else "no response"
            if a == "get":
                if not key:
                    raise ValueError("key is required")
                val = client.get(key)
                return "not found" if val is None else str(val)
            if a == "set":
                if not key:
                    raise ValueError("key is required")
                if int(ttl) > 0:
                    client.set(key, value, ex=int(ttl))
                else:
                    client.set(key, value)
                return f"set {key}"
            if a == "delete":
                if not key:
                    raise ValueError("key is required")
                return f"deleted {key}" if client.delete(key) else "not found"
            if a == "exists":
                if not key:
                    raise ValueError("key is required")
                return "yes" if client.exists(key) else "no"
            if a == "list":
                keys = [
                    k
                    for k in client.scan_iter(match=pattern, count=100)
                    if fnmatch.fnmatchcase(k, pattern)
                ]
                return "\n".join(sorted(keys)) if keys else "no keys"
            if a == "ttl":
                if not key:
                    raise ValueError("key is required")
                return str(client.ttl(key))
            if a == "expire":
                if not key:
                    raise ValueError("key is required")
                client.expire(key, int(ttl))
                return f"expire {key} {ttl}s"
            if a == "incr":
                if not key:
                    raise ValueError("key is required")
                return str(client.incr(key))
            if a == "rpush":
                if not key:
                    raise ValueError("key is required")
                return str(client.rpush(key, value))
            if a == "lrange":
                if not key:
                    raise ValueError("key is required")
                items = client.lrange(key, int(start), int(stop))
                return "\n".join(str(i) for i in items) if items else "empty"
            if a == "sadd":
                if not key:
                    raise ValueError("key is required")
                return str(client.sadd(key, member))
            if a == "smembers":
                if not key:
                    raise ValueError("key is required")
                members = client.smembers(key)
                return (
                    "\n".join(sorted(str(m) for m in members)) if members else "empty"
                )
            if a == "hset":
                if not key or not field:
                    raise ValueError("key and field are required")
                return str(client.hset(key, field, value))
            if a == "hget":
                if not key or not field:
                    raise ValueError("key and field are required")
                val = client.hget(key, field)
                return "not found" if val is None else str(val)
            if a == "hgetall":
                if not key:
                    raise ValueError("key is required")
                data = client.hgetall(key)
                return (
                    "\n".join(f"{k}={v}" for k, v in data.items()) if data else "empty"
                )
            if a == "publish":
                if not channel:
                    raise ValueError("channel is required")
                subs = client.publish(channel, message)
                return f"published to {channel} ({subs} subscriber(s))"
            raise ValueError(f"unknown action: {action}")
        finally:
            client.close()

S3GetTool

Bases: S3Tool

Download an object from an S3 bucket and return its contents.

Source code in teff/tool/builtin/s3.py

class S3GetTool(S3Tool):
    """Download an object from an S3 bucket and return its contents."""

    name = "s3_get"
    description = "Download an object from an S3 bucket and return its contents"

    def run(self, key: str = "", max_chars: int = 50000) -> str:  # type: ignore[override]
        if not self.bucket:
            raise ValueError("s3_get requires 'bucket' in config")
        if not key:
            raise ValueError("key is required")
        body = self._client().get_object(Bucket=self.bucket, Key=key)["Body"]
        return body.read(max_chars).decode("utf-8", errors="replace")

S3PutTool

Bases: S3Tool

Upload text content to an object in an S3 bucket.

Source code in teff/tool/builtin/s3.py

class S3PutTool(S3Tool):
    """Upload text content to an object in an S3 bucket."""

    name = "s3_put"
    description = "Upload text content to an object in an S3 bucket"

    def run(self, key: str = "", content: str = "") -> str:  # type: ignore[override]
        if not self.bucket:
            raise ValueError("s3_put requires 'bucket' in config")
        if not key:
            raise ValueError("key is required")
        self._client().put_object(Bucket=self.bucket, Key=key, Body=content.encode())
        return f"uploaded {len(content)} bytes to {key}"

S3Tool

Bases: Tool

List objects in an S3 bucket.

Requires boto3 (from teff[tools]). Credentials are resolved by boto3's standard chain (env vars, ~/.aws/credentials, IAM role) unless overridden via config.

Parameters:

Name	Type	Description	Default
config	dict | None	Optional dict with bucket, region, endpoint_url (for S3-compatible stores like MinIO), aws_access_key_id, aws_secret_access_key, verify (TLS verification: True/False or a CA bundle path; set False for self-signed local endpoints).	None

Source code in teff/tool/builtin/s3.py

class S3Tool(Tool):
    """List objects in an S3 bucket.

    Requires ``boto3`` (from ``teff[tools]``). Credentials are resolved by
    boto3's standard chain (env vars, ``~/.aws/credentials``, IAM role)
    unless overridden via *config*.

    Args:
        config: Optional dict with ``bucket``, ``region``,
            ``endpoint_url`` (for S3-compatible stores like MinIO),
            ``aws_access_key_id``, ``aws_secret_access_key``, ``verify``
            (TLS verification: ``True``/``False`` or a CA bundle path;
            set ``False`` for self-signed local endpoints).
    """

    name = "s3_list"
    description = "List objects in an S3 bucket"

    def __init__(self, config: dict | None = None):
        cfg = config or {}
        self.bucket = cfg.get("bucket", "")
        self.region = cfg.get("region")
        self.endpoint_url = cfg.get("endpoint_url")
        self.aws_access_key_id = cfg.get("aws_access_key_id")
        self.aws_secret_access_key = cfg.get("aws_secret_access_key")
        self.verify = cfg.get("verify", True)

    def _client(self):
        try:
            import boto3
        except ImportError as e:
            msg = "s3 tools require 'boto3' (pip install teff[tools])"
            raise ImportError(msg) from e
        kwargs = {}
        if self.region:
            kwargs["region_name"] = self.region
        if self.endpoint_url:
            kwargs["endpoint_url"] = self.endpoint_url
        if self.aws_access_key_id:
            kwargs["aws_access_key_id"] = self.aws_access_key_id
        if self.aws_secret_access_key:
            kwargs["aws_secret_access_key"] = self.aws_secret_access_key
        kwargs["verify"] = self.verify
        return boto3.client("s3", **kwargs)

    def run(self, prefix: str = "", limit: int = 100) -> str:  # type: ignore[override]
        if not self.bucket:
            raise ValueError("s3_list requires 'bucket' in config")
        response = self._client().list_objects_v2(
            Bucket=self.bucket, Prefix=prefix, MaxKeys=limit
        )
        keys = [obj["Key"] for obj in response.get("Contents", [])]
        return "\n".join(keys) if keys else "no objects found"

SQLDescribeTool

Bases: _SQLBase

Describe a table's columns and types.

Source code in teff/tool/builtin/sql.py

class SQLDescribeTool(_SQLBase):
    """Describe a table's columns and types."""

    name = "sql_describe"
    description = "Describe a table's columns and types"

    def run(self, table: str = "") -> str:  # type: ignore[override]
        if not table:
            raise ValueError("table is required")
        conn = self._connect()
        try:
            if self.db_type == "sqlite":
                cursor = conn.execute(f'PRAGMA table_info("{table}")')
                return self._format(cursor.description, cursor.fetchall())
            with conn.cursor() as cursor:
                cursor.execute(
                    "SELECT column_name, data_type, is_nullable "
                    "FROM information_schema.columns "
                    "WHERE table_schema='public' AND table_name=%s "
                    "ORDER BY ordinal_position",
                    (table,),
                )
                return self._format(cursor.description, cursor.fetchall())
        finally:
            conn.close()

SQLListTablesTool

Bases: _SQLBase

List the tables in a database.

Source code in teff/tool/builtin/sql.py

class SQLListTablesTool(_SQLBase):
    """List the tables in a database."""

    name = "sql_list_tables"
    description = "List the tables in a database"

    def run(self) -> str:  # type: ignore[override]
        conn = self._connect()
        try:
            if self.db_type == "sqlite":
                cursor = conn.execute(_SQLITE_TABLES)
            else:
                with conn.cursor() as cursor:
                    cursor.execute(_POSTGRES_TABLES)
                    return self._format(cursor.description, cursor.fetchall())
            return self._format(cursor.description, cursor.fetchall())
        finally:
            conn.close()

SQLQueryTool

Bases: _SQLBase

Run a read-only SQL query against a database.

Only SELECT/WITH (read) statements are allowed; anything that would mutate data (INSERT, UPDATE, DELETE, DDL, …) is rejected. Placeholders match the backend: ? for SQLite, %s for PostgreSQL.

Source code in teff/tool/builtin/sql.py

class SQLQueryTool(_SQLBase):
    """Run a read-only SQL query against a database.

    Only ``SELECT``/``WITH`` (read) statements are allowed; anything that
    would mutate data (``INSERT``, ``UPDATE``, ``DELETE``, DDL, …) is
    rejected. Placeholders match the backend: ``?`` for SQLite, ``%s``
    for PostgreSQL.
    """

    name = "sql_query"
    description = "Run a read-only SQL SELECT query against a database"

    def _guard(self, query: str) -> None:
        first = query.lstrip().split(None, 1)[0].upper() if query.strip() else ""
        if first not in ("SELECT", "WITH", "EXPLAIN"):
            msg = f"sql_query is read-only: unsupported statement '{first or query}'"
            raise ValueError(msg)

    def run(self, query: str = "", params: list | None = None, limit: int = 100) -> str:  # type: ignore[override]
        if not query:
            raise ValueError("query is required")
        self._guard(query)
        conn = self._connect()
        try:
            if self.db_type == "sqlite":
                cursor = conn.execute(query, params or ())
            else:
                with conn.cursor() as cursor:
                    cursor.execute(query, params or ())
                    rows = cursor.fetchmany(limit)
                    return self._format(cursor.description, rows)
            return self._format(cursor.description, cursor.fetchmany(limit))
        finally:
            conn.close()

SendEmailTool

Bases: Tool

Send an email message via SMTP.

Uses the stdlib smtplib/email modules — no extra dependency.

Parameters:

Name	Type	Description	Default
config	dict | None	Optional dict with host (SMTP server), port (default 587), username, password, from_addr (sender address), starttls (default True).	None

Source code in teff/tool/builtin/notify.py

class SendEmailTool(Tool):
    """Send an email message via SMTP.

    Uses the stdlib ``smtplib``/``email`` modules — no extra dependency.

    Args:
        config: Optional dict with ``host`` (SMTP server), ``port``
            (default 587), ``username``, ``password``, ``from_addr``
            (sender address), ``starttls`` (default True).
    """

    name = "send_email"
    description = "Send an email message via SMTP"

    def __init__(self, config: dict | None = None):
        cfg = config or {}
        self.host = cfg.get("host", "")
        self.port = cfg.get("port", 587)
        self.username = cfg.get("username", "")
        self.password = cfg.get("password", "")
        self.from_addr = cfg.get("from_addr", "")
        self.starttls = cfg.get("starttls", True)

    def run(self, to: str = "", subject: str = "", body: str = "") -> str:  # type: ignore[override]
        if not self.host:
            raise ValueError("send_email requires 'host' in config")
        if not self.from_addr:
            raise ValueError("send_email requires 'from_addr' in config")
        if not to:
            raise ValueError("to is required")

        import smtplib
        from email.mime.text import MIMEText

        message = MIMEText(body, "plain", "utf-8")
        message["Subject"] = subject
        message["From"] = self.from_addr
        message["To"] = to

        with smtplib.SMTP(self.host, self.port, timeout=30) as server:
            if self.starttls:
                server.starttls()
            if self.username:
                server.login(self.username, self.password)
            server.sendmail(self.from_addr, to, message.as_string())
        return f"email sent to {to}"

SendTelegramTool

Bases: Tool

Send a message via the Telegram Bot API.

Uses httpx (a core dependency). Config keys token (bot token) and chat_id (default recipient).

Parameters:

Name	Type	Description	Default
config	dict | None	Optional dict with token and chat_id.	None

Source code in teff/tool/builtin/notify.py

class SendTelegramTool(Tool):
    """Send a message via the Telegram Bot API.

    Uses ``httpx`` (a core dependency). Config keys ``token`` (bot token)
    and ``chat_id`` (default recipient).

    Args:
        config: Optional dict with ``token`` and ``chat_id``.
    """

    name = "send_telegram"
    description = "Send a message via a Telegram bot"

    def __init__(self, config: dict | None = None):
        cfg = config or {}
        self.token = cfg.get("token", "")
        self.chat_id = cfg.get("chat_id", "")

    async def arun(self, text: str = "", chat_id: str = "") -> str:  # type: ignore[override]
        if not self.token:
            raise ValueError("send_telegram requires 'token' in config")
        target = chat_id or self.chat_id
        if not target:
            raise ValueError("chat_id is required")
        if not text:
            raise ValueError("text is required")

        import httpx

        url = f"https://api.telegram.org/bot{self.token}/sendMessage"
        async with httpx.AsyncClient(timeout=15) as client:
            response = await client.post(url, json={"chat_id": target, "text": text})
            response.raise_for_status()
        return f"telegram message sent to {target}"

ShellTool

Bases: Tool

Run shell commands without a shell.

The command is split into an argument vector and executed via execve directly — /bin/sh is never involved, so &&, ;, pipes, backticks and $(...) are inert literal arguments, never executed. Tokens that still contain shell metacharacters (globs, redirections, quotes, whitespace) are rejected.

Parameters:

Name	Type	Description	Default
root_dir	str	Working directory for the command.	'.'
allowed_commands	list[str] | None	If set, only commands whose first token is in this list are permitted. None (default) allows all commands. A built-in blocklist of dangerous commands (sudo, dd, reboot, …) is always enforced.	None

Source code in teff/tool/builtin/shell.py

class ShellTool(Tool):
    """Run shell commands without a shell.

    The command is split into an argument vector and executed via
    ``execve`` directly — ``/bin/sh`` is never involved, so ``&&``, ``;``,
    pipes, backticks and ``$(...)`` are inert literal arguments, never
    executed.  Tokens that still contain shell metacharacters (globs,
    redirections, quotes, whitespace) are rejected.

    Args:
        root_dir: Working directory for the command.
        allowed_commands: If set, only commands whose first token is in this
            list are permitted.  ``None`` (default) allows all commands.
            A built-in blocklist of dangerous commands (``sudo``, ``dd``,
            ``reboot``, …) is always enforced.
    """

    name = "shell"
    description = "Run shell commands"

    def __init__(self, root_dir: str = ".", allowed_commands: list[str] | None = None):
        self.root_dir = root_dir
        self._allowed = allowed_commands

    async def arun(self, command: str = "") -> str:  # type: ignore[override]
        cmd = shlex.split(command)
        if not cmd:
            raise ValueError("empty command")
        prog = cmd[0]
        if prog in _DEFAULT_BLOCKED:
            raise PermissionError(f"blocked command: {prog}")
        if self._allowed is not None and prog not in self._allowed:
            raise PermissionError(
                f"command not allowed: {prog} (allowed: {self._allowed})"
            )
        for token in cmd:
            if _SHELL_METACHARS.intersection(token):
                raise PermissionError(
                    f"shell metacharacters are not allowed in: {token!r}"
                )

        proc = await asyncio.create_subprocess_exec(
            *cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            cwd=self.root_dir,
        )
        stdout, stderr = await proc.communicate()
        if proc.returncode != 0:
            msg = stderr.decode().strip()
            raise RuntimeError(msg)
        return stdout.decode().strip()

SlackSendTool

Bases: Tool

Send a message to a Slack channel.

Requires slack-sdk (from teff[tools]) and a bot token.

Parameters:

Name	Type	Description	Default
config	dict | None	Optional dict with token (bot token) and channel (default channel, e.g. #general).	None

Source code in teff/tool/builtin/slack.py

class SlackSendTool(Tool):
    """Send a message to a Slack channel.

    Requires ``slack-sdk`` (from ``teff[tools]``) and a bot token.

    Args:
        config: Optional dict with ``token`` (bot token) and
            ``channel`` (default channel, e.g. ``#general``).
    """

    name = "slack_send"
    description = "Send a message to a Slack channel"

    def __init__(self, config: dict | None = None):
        cfg = config or {}
        self.token = cfg.get("token", "")
        self.channel = cfg.get("channel", "")

    def run(self, text: str = "", channel: str = "") -> str:  # type: ignore[override]
        if not self.token:
            raise ValueError("slack_send requires 'token' in config")
        target = channel or self.channel
        if not target:
            raise ValueError("channel is required")
        try:
            from slack_sdk import WebClient
        except ImportError as e:
            msg = "slack_send requires 'slack-sdk' (pip install teff[tools])"
            raise ImportError(msg) from e

        response = WebClient(token=self.token).chat_postMessage(
            channel=target, text=text
        )
        return f"sent to {target} (ts={response['ts']})"

WaitForTool

Bases: Tool

Poll until a condition holds or a timeout elapses.

Conditions (condition):

• url — poll target (a URL) with HTTP GET until it responds; status controls what counts as success (default success = 2xx).
• redis_key — poll target (a key) in a Redis-compatible store until it exists.

Parameters:

Name	Type	Description	Default
condition		url or redis_key.	required
target		URL or key to poll.	required
timeout		Seconds before giving up (default from config, 120.0).	required
poll_interval		Seconds between checks (default from config, 1.0).	required
status		For url: success (2xx), any, or an exact HTTP status code.	required

Args (config): poll_interval, timeout, and connection keys for the redis_key condition (same as the redis tool).

Source code in teff/tool/builtin/wait_for.py

class WaitForTool(Tool):
    """Poll until a condition holds or a timeout elapses.

    Conditions (``condition``):

    - ``url`` — poll ``target`` (a URL) with HTTP GET until it responds;
      ``status`` controls what counts as success (default ``success`` =
      2xx).
    - ``redis_key`` — poll ``target`` (a key) in a Redis-compatible
      store until it exists.

    Args:
        condition: ``url`` or ``redis_key``.
        target: URL or key to poll.
        timeout: Seconds before giving up (default from config, 120.0).
        poll_interval: Seconds between checks (default from config, 1.0).
        status: For ``url``: ``success`` (2xx), ``any``, or an exact
            HTTP status code.

    Args (config): ``poll_interval``, ``timeout``, and connection keys
        for the ``redis_key`` condition (same as the ``redis`` tool).
    """

    name = "wait_for"
    description = (
        "Poll until a condition holds (URL reachable, Redis key exists) "
        "or a timeout elapses"
    )

    def __init__(self, config: dict | None = None):
        cfg = config or {}
        self.poll_interval = float(cfg.get("poll_interval", 1.0))
        self.timeout = float(cfg.get("timeout", 120.0))
        self.url = cfg.get("url", "")
        self.host = cfg.get("host", "localhost")
        self.port = cfg.get("port", 6379)
        self.db = cfg.get("db", 0)
        self.password = cfg.get("password", "")
        self.username = cfg.get("username", "")

    def _redis(self):
        try:
            import redis
        except ImportError as e:
            msg = "wait_for redis_key requires the 'redis' package (pip install teff[tools])"
            raise ImportError(msg) from e
        if self.url:
            return redis.Redis.from_url(self.url, decode_responses=True)
        kwargs: dict = {}
        if self.username:
            kwargs["username"] = self.username
        if self.password:
            kwargs["password"] = self.password
        return redis.Redis(
            host=self.host,
            port=int(self.port),
            db=int(self.db),
            decode_responses=True,
            **kwargs,
        )

    def run(  # type: ignore[override]
        self,
        condition: str,
        target: str = "",
        timeout: float | None = None,
        poll_interval: float | None = None,
        status: str = "success",
    ) -> str:
        if not condition:
            raise ValueError("condition is required (url, redis_key)")
        if not target:
            raise ValueError("target is required")
        timeout = float(timeout if timeout is not None else self.timeout)
        interval = float(
            poll_interval if poll_interval is not None else self.poll_interval
        )
        start = time.monotonic()
        if condition == "url":
            self._poll_url(target, timeout, interval, status)
        elif condition == "redis_key":
            client = self._redis()
            try:
                self._poll(lambda: bool(client.exists(target)), timeout, interval)
            finally:
                client.close()
        else:
            raise ValueError(f"unknown condition: {condition}")
        return f"condition met after {time.monotonic() - start:.1f}s"

    async def arun(  # type: ignore[override]
        self,
        condition: str,
        target: str = "",
        timeout: float | None = None,
        poll_interval: float | None = None,
        status: str = "success",
    ) -> str:
        if not condition:
            raise ValueError("condition is required (url, redis_key)")
        if not target:
            raise ValueError("target is required")
        timeout = float(timeout if timeout is not None else self.timeout)
        interval = float(
            poll_interval if poll_interval is not None else self.poll_interval
        )
        start = time.monotonic()
        if condition == "url":
            await self._poll_url_async(target, timeout, interval, status)
        elif condition == "redis_key":
            client = self._redis()
            try:
                await self._poll_async(
                    lambda: bool(client.exists(target)), timeout, interval
                )
            finally:
                client.close()
        else:
            raise ValueError(f"unknown condition: {condition}")
        return f"condition met after {time.monotonic() - start:.1f}s"

    def _poll(self, check, timeout: float, interval: float) -> None:
        deadline = time.monotonic() + timeout
        while True:
            try:
                if check():
                    return
            except Exception:
                pass
            if time.monotonic() >= deadline:
                raise ValueError(f"timed out after {timeout:.0f}s")
            time.sleep(interval)

    async def _poll_async(self, check, timeout: float, interval: float) -> None:
        import asyncio

        deadline = time.monotonic() + timeout
        while True:
            try:
                if check():
                    return
            except Exception:
                pass
            if time.monotonic() >= deadline:
                raise ValueError(f"timed out after {timeout:.0f}s")
            await asyncio.sleep(interval)

    def _poll_url(self, url: str, timeout: float, interval: float, status: str) -> None:
        if status not in ("success", "any") and not str(status).isdigit():
            raise ValueError(f"unknown status expectation: {status}")

        def check() -> bool:
            import httpx

            try:
                response = httpx.get(url, timeout=interval + 2, follow_redirects=True)
            except Exception:
                return False
            code = response.status_code
            if status == "any":
                return True
            if status == "success":
                return 200 <= code < 300
            return code == int(status)

        self._poll(check, timeout, interval)

    async def _poll_url_async(
        self, url: str, timeout: float, interval: float, status: str
    ) -> None:
        if status not in ("success", "any") and not str(status).isdigit():
            raise ValueError(f"unknown status expectation: {status}")

        import httpx

        async with httpx.AsyncClient() as client:

            async def check() -> bool:
                try:
                    response = await client.get(
                        url, timeout=interval + 2, follow_redirects=True
                    )
                except Exception:
                    return False
                code = response.status_code
                if status == "any":
                    return True
                if status == "success":
                    return 200 <= code < 300
                return code == int(status)

            await self._poll_async(check, timeout, interval)

WebFetchTool

Bases: Tool

Fetch a URL and return the page's text content.

Uses httpx (a core dependency) for the request and beautifulsoup4 (from teff[tools]) to strip markup.

Parameters:

Name	Type	Description	Default
config	dict | None	Optional dict with timeout (seconds) and user_agent.	None
timeout	float	Request timeout in seconds (default 15).	15.0
user_agent	str	User-Agent header sent with the request (default "teff").	'teff'

Source code in teff/tool/builtin/web_fetch.py

class WebFetchTool(Tool):
    """Fetch a URL and return the page's text content.

    Uses ``httpx`` (a core dependency) for the request and
    ``beautifulsoup4`` (from ``teff[tools]``) to strip markup.

    Args:
        config: Optional dict with ``timeout`` (seconds) and ``user_agent``.
        timeout: Request timeout in seconds (default 15).
        user_agent: User-Agent header sent with the request (default "teff").
    """

    name = "fetch_url"
    description = "Fetch a URL and extract its text content"

    def __init__(
        self,
        config: dict | None = None,
        *,
        timeout: float = 15.0,
        user_agent: str = "teff",
    ):
        if isinstance(config, dict):
            timeout = config.get("timeout", timeout)
            user_agent = config.get("user_agent", user_agent)
        self.timeout = timeout
        self.user_agent = user_agent

    async def arun(self, url: str = "", max_chars: int = 10000) -> str:  # type: ignore[override]
        if not url:
            raise ValueError("url is required")
        import httpx

        async with httpx.AsyncClient(
            timeout=self.timeout, follow_redirects=True
        ) as client:
            response = await client.get(url, headers={"User-Agent": self.user_agent})
            response.raise_for_status()

        try:
            from bs4 import BeautifulSoup
        except ImportError as e:
            msg = "fetch_url requires 'beautifulsoup4' (pip install teff[tools])"
            raise ImportError(msg) from e

        soup = BeautifulSoup(response.text, "html.parser")
        for tag in soup(["script", "style", "nav", "footer", "header"]):
            tag.decompose()
        text = " ".join(soup.stripped_strings).strip()
        if not text:
            return "no text content found"
        return text[:max_chars]

WebSearchTool

Bases: Tool

Search the web using DuckDuckGo (no API key required).

Source code in teff/tool/builtin/web_search.py

class WebSearchTool(Tool):
    """Search the web using DuckDuckGo (no API key required)."""

    name = "web_search"
    description = "Search the web"

    def __init__(self, provider: str = "duckduckgo"):
        self.provider = provider

    async def arun(self, query: str = "", num_results: int = 5) -> str:  # type: ignore[override]
        if self.provider == "duckduckgo":
            return await self._duckduckgo(query, num_results)
        msg = f"unknown web search provider: {self.provider}"
        raise ValueError(msg)

    async def _duckduckgo(self, query: str, num_results: int) -> str:
        url = "https://lite.duckduckgo.com/lite/"
        async with httpx.AsyncClient(timeout=15) as client:
            response = await client.post(url, data={"q": query})
            response.raise_for_status()

        import html

        text = html.unescape(response.text)
        lines = []
        count = 0
        in_link = False
        for part in text.split("<"):
            if part.startswith("a "):
                in_link = True
                continue
            if part.startswith("/a"):
                in_link = False
                continue
            if in_link:
                if ">" in part:
                    content = part.split(">", 1)[1]
                    content = content.strip()
                    if content and not content.startswith("<"):
                        lines.append(content)
                        count += 1
                        if count >= num_results:
                            break

        return "\n".join(lines) if lines else "no results"

WriteFileTool

Bases: Tool

Write text content to a file.

Source code in teff/tool/builtin/file.py

class WriteFileTool(Tool):
    """Write text content to a file."""

    name = "write_file"
    description = "Write content to a file"

    def run(self, path: str = "", content: str = "") -> str:  # type: ignore[override]
        with open(path, "w") as f:
            f.write(content)
        return f"written {len(content)} bytes to {path}"

YamlParseTool

Bases: Tool

Parse a YAML string and dump it as pretty JSON.

Source code in teff/tool/builtin/data.py

class YamlParseTool(Tool):
    """Parse a YAML string and dump it as pretty JSON."""

    name = "yaml_parse"
    description = "Parse a YAML string and dump it as JSON"

    def run(self, text: str = "", indent: int = 2) -> str:  # type: ignore[override]
        if not text:
            raise ValueError("text is required")
        try:
            import yaml
        except ImportError as e:
            msg = "yaml_parse requires 'pyyaml' (a core dependency)"
            raise ImportError(msg) from e
        data = yaml.safe_load(text)
        return json.dumps(data, ensure_ascii=False, indent=indent)