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]`
`extras_keys`	`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: TemplatedText | None = None,
    user_message: TemplatedText | None = None,
    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) 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`	`TemplatedText \| None`	`None`	—
`user_message`	`TemplatedText \| None`	`None`	—
`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`	`{}`	—

append_thread_messages

tai42_contract.agent.base.Agent.append_thread_messages

async append_thread_messages(self, *, thread_id: str, messages: list[dict[str, str]], **kwargs: Any) -> None

Append messages to thread_id’s stored history WITHOUT running the agent. Each item is {"role": "user"|"assistant", "content": str}. The conversation bridge calls this to record a manual-mode inbound and an operator’s outbound reply into the thread’s memory, so a later agent turn reads them as prior context. An agent with no thread memory raises NotImplementedError. Parameters

Parameter	Type	Default	Description
`thread_id`	`str`	—	—
`messages`	`list[dict[str, str]]`	—	—
`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`	—

AsksFinal

tai42_contract.agent.events.AsksFinal

class AsksFinal(StreamEvent)

The agent run asked its CALLER one or more questions before it could answer. Terminal. A to="caller" ask is addressed to the run’s own caller — here the SSE client driving the stream — so the run cannot proceed until the caller answers it, unlike a to="user" park (SuspendedFinal) whose answer arrives out of band. asks carries the full parked entries the caller must answer (id, question, answer format, and the rest of the parked-entry shape); the vocabulary is generic — it names no driver, engine, or resume state. Attributes

Attribute	Type
`type`	`Literal['asks_final']`
`asks`	`list[dict[str, Any]]`
`final`	`bool`

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 base tool that interprets its fixed kwargs as a nested document carries that document opaquely here. 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. response_format is a live pydantic class (in-process), the TemplatedText | dict authored-schema union (a JSON caller’s forced-output schema — inline, or named by stored id and rendered/parsed to its schema at the point of use), or unset. Attributes

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

SuspendedFinal

tai42_contract.agent.events.SuspendedFinal

class SuspendedFinal(StreamEvent)

A generic async park surfaced out of a paused agent graph. Terminal. The run parked on one or more async ask questions and will resume out of band when their answers (or expiries) arrive — it did NOT fail and did NOT stop for a live decision, so a non-streaming drain returns a suspended RECEIPT for it rather than raising AgentInterruptedError. interaction_ids are the parked questions the resume converges on; caller_interaction_ids is the subset of them addressed to the CALLER (to="caller"), empty for a park of only user asks — the two id lists let the platform’s visit normalise a caller-ask park inside an agent into its caller/user partition; expiry_at is the earliest park deadline (ISO-8601, or None when no park carried one); thread_id is the parked run’s thread. The vocabulary is generic — it names no driver, engine, or resume state, only the parked interaction ids the flow-blind platform already holds. Attributes

Attribute	Type
`type`	`Literal['suspended_final']`
`interaction_ids`	`list[str]`
`thread_id`	`str`
`caller_interaction_ids`	`list[str]`
`expiry_at`	`str \| None`
`final`	`bool`

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`

agent_session_thread

tai42_contract.agent.session_thread.agent_session_thread

agent_session_thread(thread_id: str) -> Generator[None]

Deposit thread_id as the ambient session thread for the wrapped block. The deposit is reset in a finally. A task created inside the block inherits it on a copy. Absent this wrap the deposit stays None and a run mints its own fresh thread. Parameters

Parameter	Type	Default	Description
`thread_id`	`str`	—	—

final_event_for_value

tai42_contract.agent.events.final_event_for_value

final_event_for_value(value: Any) -> MessageFinal | StructuredFinal

The terminal event a plain agent result value represents. A str is a plain-text answer (MessageFinal); any other value is structured output (StructuredFinal, its data). The single rule both the non-streaming astream default and a resumed run’s plain terminal map a raw value through, so every path surfaces the same final shape. Parameters

Parameter	Type	Default	Description
`value`	`Any`	—	—

get_agent_session_thread

tai42_contract.agent.session_thread.get_agent_session_thread

get_agent_session_thread() -> str | None

The session thread deposited for the current run, or None when none is set.

reset_agent_session_thread

tai42_contract.agent.session_thread.reset_agent_session_thread

reset_agent_session_thread(token: Token[str | None]) -> None

Restore the session thread to the value captured in token. token is the return value of the matching set_agent_session_thread call. Parameters

Parameter	Type	Default	Description
`token`	`Token[str \| None]`	—	—

set_agent_session_thread

tai42_contract.agent.session_thread.set_agent_session_thread

set_agent_session_thread(thread_id: str | None) -> Token[str | None]

Bind thread_id as the current run’s session thread. Pass the returned token to reset_agent_session_thread to restore the previous value. Parameters

Parameter	Type	Default	Description
`thread_id`	`str \| None`	—	—

HTTP API

CLI

Settings

Capability map

Conversation bridge

Python SDK

Studio SDK

Agents (tai42_contract.agent)

Agent

Members

from_tool_input

run

astream

append_thread_messages

AgentInterruptedError

AsksFinal

InterruptFinal

MessageDelta

MessageFinal

PresetSpec

ReasoningStep

RunUsage

StreamEvent

StructuredFinal

SubAgentSpec

SuspendedFinal

ToolCallStep

ToolResultStep

agent_session_thread

final_event_for_value

get_agent_session_thread

reset_agent_session_thread

set_agent_session_thread

​Agent

​Members

​from_tool_input

​run

​astream

​append_thread_messages

​AgentInterruptedError

​AsksFinal

​InterruptFinal

​MessageDelta

​MessageFinal

​PresetSpec

​ReasoningStep

​RunUsage

​StreamEvent

​StructuredFinal

​SubAgentSpec

​SuspendedFinal

​ToolCallStep

​ToolResultStep

​agent_session_thread

​final_event_for_value

​get_agent_session_thread

​reset_agent_session_thread

​set_agent_session_thread

Agent

Members

from_tool_input

run

astream

append_thread_messages

AgentInterruptedError

AsksFinal

InterruptFinal

MessageDelta

MessageFinal

PresetSpec

ReasoningStep

RunUsage

StreamEvent

StructuredFinal

SubAgentSpec

SuspendedFinal

ToolCallStep

ToolResultStep

agent_session_thread

final_event_for_value

get_agent_session_thread

reset_agent_session_thread

set_agent_session_thread