> ## Documentation Index
> Fetch the complete documentation index at: https://docs.tai42.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Agents (tai42_contract.agent)

> The Agent base class plugin authors subclass.

Agent contract: the `Agent` ABC + its neutral streamed-event vocabulary.

Re-exports the `Agent` interface (and its `PresetSpec` / `SubAgentSpec`
inputs and `AgentInterruptedError` error) alongside the `StreamEvent` types an
agent's `astream` yields and `run` drains.

## Agent

`tai42_contract.agent.base.Agent`

```python theme={null}
class Agent(ABC)
```

The uniform agent contract. Implement `run`; override
`astream` only when the agent produces real per-step events.

Subclasses declare three class attributes that drive auto-tool generation:

* `tool_name` — the registered name (also the auto-generated tool name);
* `tool_description` — the LLM-facing description (carried, not lost);
* `ToolInput` — a JSON-able pydantic model of the tool params. Live
  `tools=` is deliberately NOT in `ToolInput` (API-only via `astream`);
  a non-JSON field fails loudly at `model_json_schema()` time.

`spec_runnable` is a capability marker (default `False`):
`spec_runnable = True` declares that this agent's `ToolInput` advertises
EXACTLY the composable fields its runtime honors — no more, no less. This lets
a generic UI gate authoring controls on the input schema's fields: it renders a
control only for an advertised (honored) field, and baking an un-advertised
field is rejected at author time. The declaration is the implementation's — it
is never inferred, and no agent name is ever hardcoded. Because every
`ToolInput` field of a `spec_runnable` agent is honored, all of them are
preset-bakeable; `preset_bakeable_fields` covers the other case.

`preset_bakeable_fields` names the `ToolInput` fields whose baked
(preset `fixed_kwargs`) value the runtime honors on an agent that is NOT
UI-composable — declaring a field here implies no UI control at all, only that
a hidden baked value is passed through and acted on. A field is preset-bakeable
iff the agent is `spec_runnable` (all fields, above) OR the field is listed
here. Like `spec_runnable` it is the implementation's own declaration, never
inferred; every listed name must be a real `ToolInput` field. The empty
default means nothing is bakeable unless the agent is `spec_runnable`.

**Attributes**

| Attribute                | Type              |
| ------------------------ | ----------------- |
| `tool_name`              | `str`             |
| `tool_description`       | `str`             |
| `ToolInput`              | `type[BaseModel]` |
| `spec_runnable`          | `bool`            |
| `preset_bakeable_fields` | `frozenset[str]`  |

### Members

#### from\_tool\_input

`tai42_contract.agent.base.Agent.from_tool_input`

```python theme={null}
from_tool_input(cls, validated: BaseModel) -> dict[str, Any]
```

Map a validated `ToolInput` instance to `run` kwargs.

Passes through every set field with nested pydantic values (`presets`,
`subagents`, `inline_skills`) kept as model instances, since `run`
and the resolvers it calls read them by attribute. `model_dump` is
deliberately NOT used — it would flatten those nested models to dicts.
Agents whose JSON tool-face shape differs from their `run` signature
override this.

Any error an override raises must be a constant, hand-authored message and
must NOT interpolate input values: the agent run routes surface these
messages verbatim to the caller, and the baked input can carry credentials
the caller is not entitled to see.

**Parameters**

| Parameter   | Type        | Default | Description |
| ----------- | ----------- | ------- | ----------- |
| `validated` | `BaseModel` | —       | —           |

#### run

`tai42_contract.agent.base.Agent.run`

```python theme={null}
async run(
    self,
    *,
    tools: Sequence[StructuredTool] = (),
    tool_names: Sequence[str] = (),
    presets: Sequence[PresetSpec] = (),
    subagents: Sequence[SubAgentSpec] = (),
    system_message: str = '',
    user_message: str = '',
    response_format: Any = None,
    strategy: str | None = None,
    interrupt_on: dict[str, Any] | None = None,
    skills: Sequence[str] = (),
    inline_skills: Sequence[dict[str, Any]] = (),
    recursion_limit: int | None = None,
    thread_id: str | None = None,
    resume: Any = None,
    resume_checkpoint_id: str | None = None,
    llm_provider: str | None = None,
    checkpoint_provider: str | None = None,
    store_provider: str | None = None,
    llm_kwargs: dict[str, Any] | None = None,
    **kwargs: Any,
) -> Any
```

Run the agent once and return its final value.

Tool inputs (`tools`/`tool_names`/`presets`) are uniform only for
tools-agent-shaped agents; role-specific agents (voting/vqa/refine/
mcp\_tools) read their own params off `**kwargs`. `response_format` is
typed `Any` because the deep path passes a JSON-Schema `dict` while
the tools-agent path passes a pydantic class. `resume_checkpoint_id`
forks past an aborted turn, distinct from `resume` (the interrupt
answer).

**Parameters**

| Parameter              | Type                       | Default | Description |
| ---------------------- | -------------------------- | ------- | ----------- |
| `tools`                | `Sequence[StructuredTool]` | `()`    | —           |
| `tool_names`           | `Sequence[str]`            | `()`    | —           |
| `presets`              | `Sequence[PresetSpec]`     | `()`    | —           |
| `subagents`            | `Sequence[SubAgentSpec]`   | `()`    | —           |
| `system_message`       | `str`                      | `''`    | —           |
| `user_message`         | `str`                      | `''`    | —           |
| `response_format`      | `Any`                      | `None`  | —           |
| `strategy`             | `str \| None`              | `None`  | —           |
| `interrupt_on`         | `dict[str, Any] \| None`   | `None`  | —           |
| `skills`               | `Sequence[str]`            | `()`    | —           |
| `inline_skills`        | `Sequence[dict[str, Any]]` | `()`    | —           |
| `recursion_limit`      | `int \| None`              | `None`  | —           |
| `thread_id`            | `str \| None`              | `None`  | —           |
| `resume`               | `Any`                      | `None`  | —           |
| `resume_checkpoint_id` | `str \| None`              | `None`  | —           |
| `llm_provider`         | `str \| None`              | `None`  | —           |
| `checkpoint_provider`  | `str \| None`              | `None`  | —           |
| `store_provider`       | `str \| None`              | `None`  | —           |
| `llm_kwargs`           | `dict[str, Any] \| None`   | `None`  | —           |
| `kwargs`               | `Any`                      | `{}`    | —           |

#### astream

`tai42_contract.agent.base.Agent.astream`

```python theme={null}
async astream(self, **kwargs: Any) -> AsyncIterator[StreamEvent]
```

Free default for non-streaming agents: run once, emit one terminal.

The terminal type matches the result: a `str` becomes a
`MessageFinal` (plain-text answer), any other value a
`StructuredFinal` (the structured `response_format` object).
Streaming agents override this with the real per-step projection and
implement `run` by draining their own `astream` via
`_drain`.

**Parameters**

| Parameter | Type  | Default | Description |
| --------- | ----- | ------- | ----------- |
| `kwargs`  | `Any` | `{}`    | —           |

## AgentInterruptedError

`tai42_contract.agent.base.AgentInterruptedError`

```python theme={null}
class AgentInterruptedError(Exception)
```

Raised by `run`/`_drain` when a non-streaming caller drains a run that
paused on an interrupt. A non-streaming caller cannot act on an interrupt, so
it surfaces loudly rather than returning a partial. `interrupts` is the list
of `InterruptFinal` events the run emitted.

**Attributes**

| Attribute    | Type |
| ------------ | ---- |
| `interrupts` | —    |

## InterruptFinal

`tai42_contract.agent.events.InterruptFinal`

```python theme={null}
class InterruptFinal(StreamEvent)
```

A platform interrupt surfaced out of a paused agent graph. Terminal.

`interrupt_id` is the graph interrupt id (echoed back when resuming);
`payload` is the interrupt value the agent raised (e.g. the option set);
`reason` is an optional human-facing label. This event is emitted verbatim
as one SSE frame (`type: "interrupt_final"`); a consumer reads
`interrupt_id`, `payload` and `reason` off it — do NOT rename `payload`
or drop `reason`.

**Attributes**

| Attribute      | Type                         |
| -------------- | ---------------------------- |
| `type`         | `Literal['interrupt_final']` |
| `interrupt_id` | `str`                        |
| `payload`      | `Any`                        |
| `reason`       | `str \| None`                |
| `final`        | `bool`                       |

## MessageDelta

`tai42_contract.agent.events.MessageDelta`

```python theme={null}
class MessageDelta(StreamEvent)
```

A token-level chunk of the agent's final answer as it streams.

**Attributes**

| Attribute | Type                       |
| --------- | -------------------------- |
| `type`    | `Literal['message_delta']` |
| `text`    | `str`                      |

## MessageFinal

`tai42_contract.agent.events.MessageFinal`

```python theme={null}
class MessageFinal(StreamEvent)
```

The agent's complete final answer, assembled from the deltas. Terminal.

**Attributes**

| Attribute | Type                       |
| --------- | -------------------------- |
| `type`    | `Literal['message_final']` |
| `text`    | `str`                      |
| `final`   | `bool`                     |

## PresetSpec

`tai42_contract.agent.base.PresetSpec`

```python theme={null}
class PresetSpec(BaseModel)
```

A base tool bound to fixed kwargs, resolved into a `StructuredTool` at
run time (see `resolve_tools`). A sub-flow is `base_tool="flow"` with
`fixed_kwargs={"flow_graph": ...}`.

**Attributes**

| Attribute      | Type             |
| -------------- | ---------------- |
| `name`         | `str`            |
| `description`  | `str`            |
| `base_tool`    | `str`            |
| `fixed_kwargs` | `dict[str, Any]` |

## ReasoningStep

`tai42_contract.agent.events.ReasoningStep`

```python theme={null}
class ReasoningStep(StreamEvent)
```

A chunk of the model's intermediate reasoning ("thinking") for one step
of the agent loop. Empty/whitespace-only reasoning is never emitted.

**Attributes**

| Attribute | Type                        |
| --------- | --------------------------- |
| `type`    | `Literal['reasoning_step']` |
| `text`    | `str`                       |

## RunUsage

`tai42_contract.agent.events.RunUsage`

```python theme={null}
class RunUsage(StreamEvent)
```

Token usage (and model label) for the run. Any field may be `None`.

**Attributes**

| Attribute       | Type                   |
| --------------- | ---------------------- |
| `type`          | `Literal['run_usage']` |
| `input_tokens`  | `int \| None`          |
| `output_tokens` | `int \| None`          |
| `total_tokens`  | `int \| None`          |
| `model`         | `str \| None`          |

## StreamEvent

`tai42_contract.agent.events.StreamEvent`

```python theme={null}
class StreamEvent(BaseModel)
```

Base class for every event an `Agent` streams. `type` is a stable
discriminator; `final` marks a terminal event (see the terminal rule in
`Agent._drain`).

**Attributes**

| Attribute      | Type   |
| -------------- | ------ |
| `model_config` | —      |
| `type`         | `str`  |
| `final`        | `bool` |

## StructuredFinal

`tai42_contract.agent.events.StructuredFinal`

```python theme={null}
class StructuredFinal(StreamEvent)
```

The agent's structured (non-text) output. `data` is the object the
agent produced — typically the validated `response_format` instance. A
plain-text answer is carried by `MessageFinal` instead. Terminal.

**Attributes**

| Attribute | Type                          |
| --------- | ----------------------------- |
| `type`    | `Literal['structured_final']` |
| `data`    | `Any`                         |
| `final`   | `bool`                        |

## SubAgentSpec

`tai42_contract.agent.base.SubAgentSpec`

```python theme={null}
class SubAgentSpec(BaseModel)
```

Neutral sub-agent descriptor — the in-process, live-tools shape.

Carries name/description/system\_prompt/tool\_names/tools(live)/presets/skills/
inline\_skills/response\_format/strategy/subagents — the fields a live sub-agent
(e.g. the mcp-finder) needs. `inline_skills` items are plain dicts
(`{"name", "content"}`). JSON callers use the richer `DeepSubAgentSpec`
(full field set) instead of this type.

`tools` is `list[Any]` in the contract (live `StructuredTool` objects in
the impl) — a vendor type cannot be a runtime pydantic field here.

**Attributes**

| Attribute         | Type                                        |
| ----------------- | ------------------------------------------- |
| `model_config`    | —                                           |
| `name`            | `str`                                       |
| `description`     | `str`                                       |
| `system_prompt`   | `str`                                       |
| `tool_names`      | `list[str]`                                 |
| `tools`           | `list[Any]`                                 |
| `presets`         | `list[PresetSpec]`                          |
| `skills`          | `list[str]`                                 |
| `inline_skills`   | `list[dict[str, Any]]`                      |
| `response_format` | `type[BaseModel] \| dict[str, Any] \| None` |
| `strategy`        | `str \| None`                               |
| `subagents`       | `list[SubAgentSpec]`                        |

## ToolCallStep

`tai42_contract.agent.events.ToolCallStep`

```python theme={null}
class ToolCallStep(StreamEvent)
```

One tool invocation the agent decided to make. `call_id` is the
model-assigned id the matching `ToolResultStep` carries.

**Attributes**

| Attribute | Type                        |
| --------- | --------------------------- |
| `type`    | `Literal['tool_call_step']` |
| `tool`    | `str`                       |
| `args`    | `dict[str, Any]`            |
| `call_id` | `str`                       |

## ToolResultStep

`tai42_contract.agent.events.ToolResultStep`

```python theme={null}
class ToolResultStep(StreamEvent)
```

The value one tool call returned, matched to a `ToolCallStep` by
`call_id`. `result` is passed through untouched.

**Attributes**

| Attribute  | Type                          |
| ---------- | ----------------------------- |
| `type`     | `Literal['tool_result_step']` |
| `tool`     | `str`                         |
| `call_id`  | `str`                         |
| `result`   | `Any`                         |
| `is_error` | `bool`                        |