2603 Tool Recipes B8ec6ce4

# Tool Recipes (Minimal)

Alloy emphasizes small, composable primitives. You can wire any capability as a
`@tool` using the libraries you already trust. Below are minimal, copy‑pasteable
examples you can drop into your project.

## HTTP fetch (GET)

```python
from __future__ import annotations

import urllib.request
import urllib.parse
from typing import Any
from alloy import tool

@tool
def http_fetch(url: str, *, timeout_s: float = 8.0, max_bytes: int = 200_000) -> str:
    """Fetch a URL over http/https and return text (best‑effort decode).

    Minimal design: stdlib only, timeouts, byte cap, scheme allowlist.
    """
    parsed = urllib.parse.urlparse(url)
    if parsed.scheme not in ("http", "https"):
        raise ValueError("only http/https allowed")
    req = urllib.request.Request(url, headers={"User-Agent": "alloy-tool/1"})
    with urllib.request.urlopen(req, timeout=timeout_s) as resp:  # nosec B310
        data = resp.read(max_bytes)
        try:
            return data.decode("utf-8", errors="replace")
        except Exception:
            return data.decode("latin-1", errors="replace")
```

## Local file search

```python
from __future__ import annotations

import os, glob
from alloy import tool

@tool
def file_search(query: str, *, paths: list[str], max_files: int = 5, max_chars: int = 240) -> list[dict]:
    """Search local text files (simple substring match) and return [{path, snippet}]."""
    if not query.strip() or not paths:
        raise ValueError("query and paths are required")
    patterns: list[str] = []
    for p in paths:
        patterns.append(os.path.join(p, "**", "*") if os.path.isdir(p) else p)
    hits: list[dict] = []
    needle = query.lower()
    seen: set[str] = set()
    for pat in patterns:
        for fp in glob.iglob(pat, recursive=True):
            if len(hits) >= max_files or fp in seen or not os.path.isfile(fp):
                continue
            seen.add(fp)
            if os.path.splitext(fp)[1].lower() in {".txt", ".md", ".py", ".rst", ".json"}:
                try:
                    with open(fp, "r", encoding="utf-8", errors="ignore") as f:
                        data = f.read()
                except Exception:
                    continue
                where = data.lower().find(needle)
                if where != -1:
                    start = max(0, where - max_chars // 4)
                    end = min(len(data), where + len(query) + max_chars // 2)
                    hits.append({"path": fp, "snippet": data[start:end].strip()})
    return hits
```

## Python execution (dev‑only; dangerous)

!!! danger "NEVER enable in production"
    These examples execute arbitrary Python code. Even with guards, this is **not safe** for untrusted input. For sandboxed demos only.

You can gate them behind an env var like `ALLOY_ENABLE_PY_EXEC=1`.

```python
from __future__ import annotations

import io, re, math, os, contextlib
from typing import Any
from alloy import tool

ALLOY_ENABLE_PY_EXEC = os.getenv("ALLOY_ENABLE_PY_EXEC") == "1"

def _sandbox_globals(capture: io.StringIO | None = None) -> dict[str, Any]:
    allowed = {
        "__builtins__": {"abs": abs, "min": min, "max": max, "sum": sum, "len": len, "range": range},
        "math": math,
    }
    if capture is not None:
        def _p(*args: Any, **kwargs: Any) -> None:
            sep = kwargs.get("sep", " ")
            end = kwargs.get("end", "\n")
            capture.write(sep.join(str(a) for a in args) + end)
        allowed["print"] = _p
    return allowed

@tool
def py_eval_dev(expr: str) -> str:
    if not ALLOY_ENABLE_PY_EXEC:
        raise PermissionError("py_eval_dev disabled; set ALLOY_ENABLE_PY_EXEC=1 to enable")
    if re.search(r"__|import|open|exec|eval|os\.|sys\.", expr):
        raise ValueError("disallowed tokens in expr")
    return str(eval(expr, _sandbox_globals(), {}))

@tool
def py_exec_dev(code: str) -> str:
    if not ALLOY_ENABLE_PY_EXEC:
        raise PermissionError("py_exec_dev disabled; set ALLOY_ENABLE_PY_EXEC=1 to enable")
    if re.search(r"__|import|open|exec\(|eval\(|os\.|sys\.", code):
        raise ValueError("disallowed tokens in code")
    buf = io.StringIO()
    with contextlib.redirect_stdout(buf):
        exec(code, _sandbox_globals(capture=buf), {})
    return buf.getvalue()
```

Guidance
- Keep tools tiny; add deps only when you must.
- Validate inputs and cap work (timeouts, byte limits).
- Prefer returning compact, easy‑to‑parse shapes from tools.
- If a tool grows complex, consider isolating it as a separate package.

## Provider‑backed web search (optional)

Option A: SerpAPI (requires `requests` and `SERPAPI_KEY`)

```python
from __future__ import annotations

import os, requests
from alloy import tool

SERPAPI_KEY = os.getenv("SERPAPI_KEY")

@tool
def web_search_serpapi(query: str, *, max_results: int = 3) -> list[dict]:
    if not SERPAPI_KEY:
        raise RuntimeError("Set SERPAPI_KEY to enable web_search_serpapi")
    r = requests.get(
        "https://serpapi.com/search",
        params={"q": query, "engine": "google", "api_key": SERPAPI_KEY},
        timeout=10,
    )
    r.raise_for_status()
    data = r.json()
    items = []
    for item in (data.get("organic_results") or [])[:max_results]:
        items.append({
            "title": item.get("title", ""),
            "url": item.get("link", ""),
            "snippet": item.get("snippet", ""),
        })
    return items
```

Option B: DuckDuckGo (requires `duckduckgo_search`)

```python
from __future__ import annotations

from duckduckgo_search import DDGS
from alloy import tool

@tool
def web_search_ddg(query: str, *, max_results: int = 3) -> list[dict]:
    with DDGS() as ddgs:
        results = ddgs.text(query, max_results=max_results) or []
    return [{"title": r.get("title", ""), "url": r.get("href", ""), "snippet": r.get("body", "")} for r in results]
```

## Design by Contract (DBC) for tool sequences

Use `@require` and `@ensure` to guide the model with early feedback.

```python
from __future__ import annotations

from alloy import tool, require, ensure, command

@tool
@ensure(lambda d: isinstance(d, dict) and "validated_at" in d, "must add validated_at")
def validate_data(data: dict) -> dict:
    # idempotent enrichment
    data = dict(data)
    data.setdefault("validated_at", "2025-01-01T00:00:00Z")
    return data

@tool
@require(lambda ba: isinstance(ba.arguments.get("data"), dict) and "validated_at" in ba.arguments["data"],
         "run validate_data first")
@ensure(lambda ok: ok is True, "save must succeed")
def save_to_production(data: dict) -> bool:
    # pretend to save
    return True

@command(output=str, tools=[validate_data, save_to_production])
def normalize_then_save(payload: dict) -> str:
    return f"Normalize the payload using validate_data, then call save_to_production. Payload: {payload}"
```

Notes
- On a failed `@require/@ensure`, the tool returns the configured message to the model (not a hard error).
- The model can then correct course (e.g., call `validate_data` before `save_to_production`).
- Keep messages short and actionable.

Narrative
1. The model calls `save_to_production(payload)`.
2. `@require` fails because `validated_at` is missing → tool returns "run validate_data first".
3. The model adapts and calls `validate_data(payload)` → gets enriched payload.
4. The model calls `save_to_production(enriched)`; `@ensure` passes → proceed.