TAI42 — the OS for AI tools

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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`

HTTP API

CLI

Settings

Python SDK

Catalog

Studio SDK

Agents (tai42_contract.agent)

Agent

Members

from_tool_input

run

astream

AgentInterruptedError

InterruptFinal

MessageDelta

MessageFinal

PresetSpec

ReasoningStep

RunUsage

StreamEvent

StructuredFinal

SubAgentSpec

ToolCallStep

ToolResultStep

​Agent

​Members

​from_tool_input

​run

​astream

​AgentInterruptedError

​InterruptFinal

​MessageDelta

​MessageFinal

​PresetSpec

​ReasoningStep

​RunUsage

​StreamEvent

​StructuredFinal

​SubAgentSpec

​ToolCallStep

​ToolResultStep

Agent

Members

from_tool_input

run

astream

AgentInterruptedError

InterruptFinal

MessageDelta

MessageFinal

PresetSpec

ReasoningStep

RunUsage

StreamEvent

StructuredFinal

SubAgentSpec

ToolCallStep

ToolResultStep