# yait-aichain

> The provider-agnostic AI toolkit for Python: a small library for AI
> pipelines and agents, written to be built on by coding agents. Deterministic pipelines — Skill, Chain, Pool; an Agent
> whose mode follows the task — agile, waterfall, a team; tools, any REST API
> or MCP server, utilities for vectors and a working folder; cost, budgets,
> permissions, events and eval on every run. The same code runs on a laptop
> and in production, and switching provider is changing the model name.

Install: `pip install yait-aichain`. Keys come from the environment
(`OPENAI_API_KEY`, `ANTHROPIC_API_KEY`, `GOOGLE_AI_API_KEY`, …).

This is the 4.x API. Names an agent may remember from older tutorials do not
exist, and each raises with the one to use: `orchestrator=`, `max_steps=`,
`max_attempts=`, `executors=`, `memory=`, `VectorDB`, `vectorQuery`,
`Embedding(...)`, `Reranker(...)`, `LocalTools`, `MCPTools`, `TTS`, `STT`,
`convertToMD`. Tools are made by service name — `search("exa")`,
`fetch("local")` — a function becomes one with `@tool`, and an image is
`image(service, model=…)`, never a `Skill`.

## The rules an agent gets wrong

1. **An agent's limit is `stop_when`, never `max_steps`.**
   `Agent(model, tools=[...], stop_when=[step_count(10)])`; the default is
   `[step_count(30)]`. A ceiling ends the run as a failure; `check(fn)` ends it
   on a fact your code verified; `stalled(k)` / `repeating(k)` nudge the model
   and let it continue.
2. **A Skill takes `prompt=` or `input=`, not both.** `prompt="Summarise:
   {text}"` is one user message. `input={"messages": [...]}` is the full spec,
   and a message is `{"role": "system" | "user" | "assistant", "parts": [...]}`
   — **`parts`, never `content`**. `{name}` placeholders are filled from
   `run(variables={...})`.
3. **`skill.run()` returns the value; Chain, Pool and Agent return a result.**
   The three results share `.output`, `.success`, `.error`, `.history`,
   `.usage`, `.tokens_used`, `.cost`, and `bool(result)` is `.success`. A
   chain result also reads any step by its key: `result["tweet"]`. A pool
   result iterates and indexes like the list of outputs.
4. **Drive the loop yourself only if you answer every call.** After
   `agent.step()` returns a `ToolCallRequest`, append `reply.as_turn()` and
   then `tool_result_turn(call.id, result)` for **every** call — failed ones
   too, with the error as the result — or the next `step()` raises.
   `agent.run()` does this for you; prefer it.
5. **A tool is called with `input=` and `options={...}`**, and an option is
   declared under `parameters["properties"]["options"]["properties"]`. An
   undeclared option is warned about and not used.
6. **In a Chain, a Tool step receives the variables named like its
   `parameters`**; map them with a third element — a name `(tool, "count",
   {"input": "text"})`, a template `{"input": "Draw {topic}"}`, or a dict
   `{"options": {"query": "{q}", "top_n": 5}}`. An Agent step reads its task
   from the variable **`task`**.
7. **A run pauses only where a `Wait`/`Gate` asks it to**, and it needs a
   `store=`. In a Chain that is a step; for an agent it is a tool the model
   may call — `Agent(tools=[gate(publish)], store=FileStore("runs/"))` — and
   an agent given one of those tools **without** a store is refused at
   construction. Nothing pauses a run from outside it.
8. **One error vocabulary**: `raise`, `stop`, `skip` (warn), `collect`
   (silent). `Chain(on_step_error=...)` defaults to `raise`, `Pool(on_error=...)`
   to `collect`.

## Signatures

<!-- g:signatures -->
```python
Model(name: str, options: dict | None = None, client_options: dict | None = None,
      api_key: str | None = None, on_unsupported: str = 'warn')
Skill(model: Model | list[Model], input: dict | None = None, *,
      prompt: str | None = None, output: dict | None = None,
      variables: dict | None = None, name: str | None = None,
      description: str | None = None, max_retries: int = 0, retry_delay: float = 2.0,
      hooks: list | None = None, max_cost: float | Budget | None = None, **refused)
Chain(steps, variables: dict | None = None, name: str | None = None,
      description: str | None = None, on_step_error: str = 'raise', store = None,
      hooks: list | None = None, max_cost = None)
Pool(runner, items: list[dict], max_flows: int = 10, on_error: str = 'collect',
     name: str | None = None, description: str | None = None, max_cost = None)
Agent(model: Model, tools: list | None = None, instructions: str = '',
      mode: str = 'agile', team: list[Agent] | str | None = None,
      planner_model: Model | None = None, stop_when: list | None = None,
      hooks: list | None = None, permissions: PermissionPolicy | None = None,
      approve: Callable[[ApprovalRequest], Any] | None = None,
      store: StateStore | None = None, max_cost: float | Budget | None = None,
      name: str | None = None, description: str | None = None, verbose: int = 0)
```
<!-- /g:signatures -->

## Imports

```python
from yait_aichain import Agent, Budget, BudgetExceeded, Chain, Model, Pool, Skill, Tracer
from yait_aichain.agent import check, stalled, step_count, token_budget
from yait_aichain.models import ToolCallRequest, tool_result_turn
from yait_aichain.state import FileStore, RunContext, SuspendedResult
from yait_aichain.tools import ApprovalDecision, PermissionPolicy, Tool, gate, wait
```

## Model

```python
Model("claude-sonnet-4-6")                          # provider from the name
Model("claude-sonnet-4-6", options={"temperature": 0.2, "max_tokens": 800})
Model("private/Qwen/Qwen3-8B",                      # your own vLLM / Ollama server,
      client_options={"url": "http://gpu.internal:8000"})   # no API key needed
```

`client_options` takes `url`, `timeout`, `retries` and `proxy`; any other key
raises. `PRIVATE_BASE_URL` sets the address of `private/` models instead.

An option a provider has no control for is declined with a warning and
recorded on `skill.last_adaptations`; an option name that does not exist raises.

## Skill — one model call

```python
summarise = Skill(Model("claude-sonnet-4-6"), prompt="Summarise in one line: {text}")
line = summarise.run(variables={"text": "…"})                   # a string

# A system message is a standing instruction; variables= on the Skill are defaults.
brief = Skill(Model("gpt-5.4-mini"), input={"messages": [
    {"role": "system", "parts": ["Reply only in French."]},
    {"role": "user", "parts": ["What is the capital of {country}?"]},
]}, variables={"country": "Canada"})
brief.run()                                     # asks about Canada, the default
brief.run(variables={"country": "Peru"})        # asks about Peru, for this call only

extract = Skill(Model("gpt-5.4-mini"), prompt="City and date in: {text}",
                output={"format": {"type": "json_schema", "schema": {
                    "type": "object",
                    "properties": {"city": {"type": "string"}, "date": {"type": "string"}},
                    "required": ["city", "date"]}}})
data = extract.run(variables={"text": "See you in Lisbon on 3 May."})   # a dict

# Several turns in one Skill: an assistant turn with no parts means "generate here".
ideas = Skill(Model("claude-sonnet-4-6"), input={"messages": [
    {"role": "user", "parts": ["Five names for a water bottle."]},
    {"role": "assistant"},
    {"role": "user", "parts": ["Pick the best one."]},
]})
best = ideas.run()                                               # the last reply

# An image is a part with a source: a URL, or {"kind": "base64", "mime": …, "data": …}.
see = Skill(Model("gpt-5.4-mini"), input={"messages": [{"role": "user", "parts": [
    "Describe this picture in one sentence.",
    {"type": "image", "source": {"kind": "url", "url": "https://example.com/cat.jpg"}},
]}]})

for piece in summarise.stream(variables={"text": "…"}):         # text as it arrives
    print(piece, end="", flush=True)
whole = summarise.last_result                                    # the full answer afterwards
```

## Tool — a function with a schema

**Any function is a tool** with `@tool`: its name is the tool's name, the first line
of its docstring the description, its annotated parameters the schema, each
described by its `name: text` docstring line. `risk` defaults to `"write"` — say
`risk="read"` when it only reads.

```python
from yait_aichain.tools import tool

@tool(risk="read")
def word_count(text: str, unique: bool = False) -> dict:
    """Count the words in a text.

    text: the text to count.
    unique: count each word once.
    """
    words = text.split()
    return {"words": len(set(words)) if unique else len(words)}

word_count.run(text="to be or not to be")    # {"words": 6}
Agent(model, tools=[word_count])             # the model sees word_count(text, unique)
```

No docstring, `*args`, or an annotation with no JSON type (`str`, `int`, `float`,
`bool`, `list[...]`, `dict`, `Literal[...]`, `Optional[...]`) is refused when the tool
is made. A chain holding a `@tool` cannot be saved — a function is code. A subclass
is the other way, when a tool holds state or needs a schema annotations cannot say:

```python
class FirstWords(Tool):
    name        = "first_words"
    description = "Return the first N words of a text."
    parameters  = {"type": "object",
                   "properties": {"input":   {"type": "string"},
                                  "options": {"type": "object",
                                              "properties": {"n": {"type": "integer"}}}},
                   "required": ["input"]}

    def run(self, input, options=None):
        return " ".join(input.split()[: (options or {}).get("n", 3)])

result = FirstWords()(input="a b c d", options={"n": 2})   # ToolResult: .success .output .error
words  = FirstWords().run(input="a b c d")                 # the bare value; raises on error
```

Calling the tool object catches an exception and returns it as `.error`;
`.run()` lets it raise. `run()` receives every property of `parameters` as a keyword argument —
`def run(self, amount, options=None)` — and the object is called the same way,
`Refund()(amount=20)`, still returning a `ToolResult`. `run()` may return any
value; a dict stays a dict.

Built-in tools (search, fetch, decisions, speech, embeddings, reranking, chunking,
local files, vector stores, REST, MCP) are in
`docs/reference/tools.md`. A provider tool is made by its provider's name —
`speech("openai")`, `transcribe("openai")`, `embed("openai", model=…)`,
`rerank("cohere", model=…)`; `chunk(max_chars=…)`, `wait(…)` and `gate(tool, …)` take
no provider. An embedding reaches a model as a one-line summary, never as the vectors.

**Images are a tool, not a model** — a model is what can be an agent's mind:
`image("openai", model="gpt-image-1.5", output_dir="images/").run("a lighthouse",
images=[…], options={"aspect_ratio": "16:9"})` → `out["images"][0]["path"]` (or
`["base64"]` without `output_dir`), `out["cost"]`. `images=` makes it an edit. A model
reads one line — what was made and where — never the bytes. `Skill` with an image model
still works, warns, and raises in 4.0.

**Utilities** hold a connection or a place: `from yait_aichain.utilities import Vectors,
MCP, REST, Shell`. Code calls their methods directly; their tools are attributes called
with `.run` — only a tool has `.run`. Give an agent the tools, never the utility:

```python
kb = Vectors("qdrant", collection="docs", description="the product documentation",
             embedder=embed("openai", model="text-embedding-3-small"))
kb.upsert(records); kb.count()                       # methods
kb.query.run("how does caching work?", {"n": 5})     # its tool
sh = Shell("projects/demo")                          # browse, read, write, run
api = REST("https://api.example.com", description="What the API is")  # api.endpoint(name, method, path, description=…)
Agent(model, tools=[kb.query, sh.read, *MCP("https://mcp.deepwiki.com/mcp").tools])
Chain(steps=[(chunk(max_chars=800), "pieces", {"input": "text"}),
             (kb.upsert, "count", {"records": "pieces"})])   # a method is a chain step
```

`Agent(tools=[kb])` or `[kb.upsert]` is refused, naming the utility's tools. A chain
with a method step cannot be saved. A tool's `risk` is
one of `read`, `draft`, `write`, `external`, `financial`, `destructive`,
`privileged`; a `@tool` is `write` unless it says otherwise. The library's own tools
say what they risk: `Shell.run` is `privileged`, a REST endpoint follows its method
(`GET` `read`, `DELETE` `destructive`, other writes `external`), and an MCP tool its
server's annotations — `privileged` when there are none. With
`permissions=PermissionPolicy()` the shipped rules allow `read`/`draft`/`write`, ask
`approve=` for `external`/`financial`/`privileged`, and deny `destructive`; with no
policy every tool runs. In 4.0.0 the policy is consulted in the agent's own loop only —
the `pool` tool, `execute()`, `resume()` and `Chain`/`Pool` steps do not consult it. A tool may keep state on its instance and define its own `__init__`.

**Search, fetch and decide — a tool made by its provider's name.** `search("exa")`,
`fetch("firecrawl")`, `decide("openai", questions=…)`. `run()` returns a `Record`: a
dict for code, whose `str()` is what a model reads — an agent shows the model that
text, a Skill prompt that substitutes the result gets it, and a Chain keeps the
record under its step's key. A paid call has `cost` (USD), `cost_basis` and `usage`.

**A decision with probabilities, not prose** — `decide("openai", …)` (OpenAI's
Decisions API, `OPENAI_API_KEY`, takes images) or `decide("typesafe", …)` (TypeSafe's
Jev, `TYPESAFE_API_KEY`). Same arguments, same answers. They write no text: they answer
questions fixed when the tool is made, about the text passed to `run()`.

```python
from yait_aichain.tools import decide

triage = decide("openai", name="triage", questions={
    "urgent": {"type": "predicate", "instructions": "Does this convey urgency?"},
    "team":   {"type": "choice", "instructions": "Which team should handle it?",
               "choices": {"billing": "Payments, refunds", "technical": "Bugs, outages"}},
    "anger":  {"type": "score", "instructions": "How frustrated is the customer?",
               "levels": ["Calm", "Frustrated", "Very angry"]},
})
out = triage.run("Help! My payouts have been failing for 3 days.")
out["answers"]["urgent"]["probability"]    # 0.96 — a probability, not a bool
out["answers"]["team"]["choice"]           # "billing"; also "probabilities", "confidence"
out["answers"]["anger"]["score"]           # 1.0 on 0..2; "probabilities" keyed by label
out["model"], out["cost"]                  # the version that answered; USD, input only
triage.run("Is this damaged?", images=["photo.png"])   # openai only
Agent(model, tools=[triage])               # the agent calls triage(state) by itself
```

Thresholds are yours: compare `probability` or `confidence` to a number in code.

**Web search, one format** — `search(service)` with `perplexity`, `brave`, `exa`,
`firecrawl`, `tavily` or `parallel`: the same options and the same records; a provider
that cannot honour an option says so in `notices` and in a `Note:` line, never silently.

```python
from yait_aichain.tools import search

found = search("perplexity").run("AI and leadership", {"max_results": 5, "domains": ["hbr.org"],
                                 "start_date": "2026-09-28", "end_date": "2026-10-04"})
found["results"][0]      # {"title", "url", "date", "author", "snippets", "text"}
found["notices"]         # options not applied here, and why
found["cost"]            # USD; found["cost_basis"] says how, found["usage"] the provider's units
str(found)               # the numbered list a model reads
```

Options: `max_results`, `recency` (hour/day/week/month/year), `start_date`, `end_date`
(YYYY-MM-DD), `country`, `language`, `domains`. A result's `date` is the provider's
claim; a fetched page carries the date it declares.

**A page from its link, one format** — `fetch(service)` with `firecrawl`, `tavily`,
`exa`, `parallel` (each with its search tool's key) or `local` (MarkItDown on this
machine, free, respects robots.txt; needs `pip install "yait-aichain[fetch]"`).

```python
from yait_aichain.tools import fetch

found = fetch("exa").run(["https://example.com/a", "https://example.com/b"], {"max_chars": 20000})
found["pages"][0]        # {"url", "final_url", "title", "published", "author", "text", "error"}
found["cost"]            # USD, as for search
str(found)               # each page as text, each failed one as its reason
```

A URL that failed is a page with `error` set, not a failed call. Search finds the
link, fetch brings the page: `search("perplexity")` + `fetch("firecrawl")` is the pair
that reached the most in our measurement (2026-10-08).

## Chain — steps in order

```python
draft  = Skill(Model("gpt-5.4-mini"), prompt="Write a tweet about {topic}.")
polish = Skill(Model("claude-sonnet-4-6"), prompt="Tighten this tweet: {tweet}")

chain = Chain(steps=[(draft, "tweet"), (polish, "final")])
result = chain.run(variables={"topic": "solar power"})   # a ChainResult
final  = result.output                                    # the last step's output
first  = result["tweet"]                                  # any step's output, by key
```

A plain dict output is merged into the variables and is not stored under its key
(as the last step, it is `result.output`); any other output — a string, a number, a
`Record` from search, fetch or decide — is stored under the key. A mapping puts
**the tool's parameter on the left and, on the right, a chain variable's name**
(`(CharCount(), "chars", {"text": "draft"})`), **a template** (`{"text": "Title:
{title}"}`, filled like a Skill prompt; `"{docs}"` alone passes the value itself) or
**a dict** whose templates are filled and other values kept. A missing variable or a
parameter the tool does not take is an error naming it. `result.history` has one record
per step; `result.success` is False when a step failed under
`on_step_error="skip"`.
`chain.save("c.yaml")` / `Chain.load("c.yaml")`.

## Pool — one runner, many items, in parallel

```python
classify = Skill(Model("gpt-5.4-mini"), prompt="Positive or negative? {review}")
pool = Pool(classify, items=[{"review": r} for r in reviews], max_flows=5)
labels = pool.run()                  # a PoolResult: outputs in item order
print(labels[0], labels.success, labels.cost)
```

The runner can be a Skill, a Chain or an Agent (items then carry `task`; one
agent serves every item, each item its own run) — each item's output is that
runner's `.output` — or a Tool. Work in parallel is a `Pool`, not threads.

With the default `on_error="collect"` a failed item does not stop the others:
its output is `None`, `result.success` is False, `result.error` names the first
failure (`"item 1: bad id"`) and `result.errors` has one error or `None` per item.

## Agent — a loop that calls tools until it is done

```python
agent = Agent(Model("claude-sonnet-4-6"), tools=[FirstWords()],
              instructions="Be exact.", stop_when=[step_count(8), stalled(3)])
result = agent.run("What are the first two words of 'the quick brown fox'?")
print(result.success, result.output, result.stopped_by, result.cost)
```

Add `check(fn, name="…")` to `stop_when` when success can be verified in code:
`fn(state)` returning True ends the run and `result.stopped_by` reads
`check:<name>`. Otherwise `stopped_by` is `answered` (the model gave its final
answer), the name of the ceiling that ended it (`step_count`, `token_budget`),
or `llm_error`.
`result.journal` records what was attempted. Each `run()` is a new
conversation: to follow up, put the earlier answer in the next task.

```python
for event in agent.stream("What are the first two words of 'a b c'?"):
    print(event.type, event.payload)   # run.started, tool_call.started, text.delta {"text": …}, run.finished
print(agent.last_result.output)        # the AgentResult, once the stream is exhausted
```

An event has `type`, `name` (the tool or model), `payload` — for `tool_call.*`
the call's `id` and `arguments`, and on `ended` its result — plus `run_id`,
`step`, `usage`, `cost` and `error`.

`mode="agile"` (the default) decides every step as it goes; `mode="waterfall"`
writes a plan first and follows it, and only a waterfall agent takes
`planner_model=Model(…)` to write that plan with a different model.

### Delegation

```python
chemist = Agent(Model("gpt-5.4-mini"), name="chemist", description="Answers chemistry questions.")
lead    = Agent(Model("claude-sonnet-4-6"), team=[chemist])   # the lead gets a `delegate` tool
answer  = lead.run("The history and the chemistry of coffee.").output
```

The lead picks a worker by its `name`; each worker keeps its own `stop_when`
(default `[step_count(30)]`).

### Permissions

```python
def approve(request):                   # request.tool, .risk, .arguments
    if request.arguments.get("amount", 0) <= 50:
        return True
    return ApprovalDecision(False, "over the limit")

agent = Agent(Model("gpt-5.4-mini"), tools=[refund_tool],
              permissions=PermissionPolicy({"financial": "approve"}), approve=approve)
```

A rule maps a risk class to `"allow"`, `"approve"` or `"deny"`. `"deny"` never
runs the tool, whatever the model asks; without `approve=`, an `"approve"` rule
refuses too. A refusal reaches the model as a tool result carrying the reason.

### Driving the loop yourself

```python
agent    = Agent(Model("gpt-5.4-mini"), tools=[FirstWords()])
state    = agent.new_state()
messages = agent.opening("First two words of 'a b c'?")
while True:
    reply = agent.step(messages, state)
    if not isinstance(reply, ToolCallRequest):
        print(reply)                               # the answer
        break
    messages.append(reply.as_turn())
    for call in reply.calls:                       # every call, failed ones too
        result, error = agent.execute(call, state)
        messages.append(tool_result_turn(call.id, error or result))
```

A `ToolCall` has `id`, `name` and `arguments`. The result you append can come
from anywhere — a person, another system — not only from `agent.execute()`.

## Pause and resume

```python
chain = Chain(steps=[(draft, "tweet"), (wait(reason="approve?"), "approval"),
                     (publish, "posted")], store=FileStore("runs/"))
result = chain.run(variables={"topic": "solar power"})
if isinstance(result, SuspendedResult):
    reason = result.awaiting["reason"]
    tweet  = result.document["variables"]["tweet"]  # everything computed so far
    # later, possibly in another process with the same store:
    posted = chain.resume(result.run_id, signal={"tweet": tweet})   # a ChainResult
```

A `signal` is merged into the variables. The chain that resumes can be a new
object — built again with the same steps and the same store.

`gate(tool)` pauses before a tool and runs it only if the signal approves:

```python
chain  = Chain(steps=[(draft, "announcement"), (gate(publish), "published")], store=FileStore("runs/"))
paused = chain.run(variables={"news": "new office hours"})
final  = chain.resume(paused.run_id, signal={"approved": False})
final.output                            # {"approved": False, "skipped": True}; publish never ran
```

Approved (`{"approved": True}`), the step's output is the tool's own output.

An **agent** pauses the same way, except that the model decides when: give it
a `Wait` or a `Gate` as an ordinary tool, plus a store.

```python
agent  = Agent(model, tools=[gate(refund, reason="Approve this refund?")],
               store=FileStore("runs/"))
result = agent.run("refund order 4417 if the policy allows it")
if isinstance(result, SuspendedResult):
    result.run_id, result.awaiting            # {"reason": ..., "resume_with": ...}
    # later, another process, same store:
    done = agent.resume(result.run_id, signal={"approved": True})  # an AgentResult
```

A run can also be parked **before it starts** — `arm()` — which is how a
subscription is expressed: a run that exists, waits, and has taken no step.

```python
armed = chain.arm(reason="waiting for an order", resume_with={"topic": "str"},
                  hint={"on": "order.created"})     # or agent.arm(...)
armed.run_id                                        # store it; that is the subscription
chain.resume(armed.run_id, signal={"topic": "solar power"})   # the signal seeds the variables
agent.resume(armed.run_id, signal={"task": "refund order 4417"})  # for an agent it is the task
```

`hint` is recorded and never acted on — the library schedules nothing; a cron,
a webhook or a person brings the signal. An agent's signal may also be a bare
string, and any other key seeds the conversation like `variables=`.

The signal completes the tool that asked for the pause — a `Gate` runs the
guarded tool when approved and skips it otherwise — and its result goes back to
the model as that call's tool result, so the run continues where it stopped.
`resume()` returns an `AgentResult`, or another `SuspendedResult` under the
**same** `run_id` if the model pauses again. A `run_id` the store no longer
holds raises `KeyError`, so a trigger that delivers twice cannot run the tail
twice. There is still no `agent.save()` and no way to stop a run from outside
it: what is written down is the conversation, and only while an answer is owed.

## Prompt cache in an agent loop

On Anthropic, `Model("claude-sonnet-4-6", options={"cache_control": "last"})`
marks the newest block of the history — a `tool_result` as readily as text — so
each turn reads the previous one back. `True` marks the second-to-last message's
text, which a tool turn does not have, and caches nothing in a loop. OpenAI and
Google cache on their own; the saving shows in `usage.cache_read_tokens`.

## Money and observability

`max_cost=0.50` on Skill, Chain, Pool or Agent refuses to *begin* a call once
that much is spent in this run — each run starts the number afresh; one
`Budget(0.50)` passed to several primitives is shared and spends down across
runs.
A Skill over budget raises `BudgetExceeded`; an Agent ends the run with
`success=False`, `stopped_by="llm_error"` and the reason in `.error`.

```python
tracer = Tracer()
agent  = Agent(Model("gpt-5.4-mini"), tools=[FirstWords()], hooks=[tracer])
agent.run("First word of 'hello world'?")
print([event.type for event in tracer.events])  # run.started, tool_call.started, …
```

## More

Docs `docs/index.md` · models and prices `docs/reference/model-registry.md` ·
options per provider `docs/reference/parameters.md` · examples `examples/README.md`
