TAI42 — the OS for AI tools

The assembled TaiApp facade. One Protocol per feature (see tai42_contract.app.facets), composed into a single TaiApp protocol that exposes them as namespaces (app.tools, app.agents, …). The app members are partitioned across the sub-protocols — each lives in exactly one, save the shared leaf names store (versioning + presets) and register/get (webhook_verifiers + channels). The runtime forwarding handle is tai42_app (see tai42_contract.app.handle).

AppAdmin

tai42_contract.app.facets.AppAdmin

class AppAdmin(Protocol)

Attributes

Attribute	Type
`live_manifest` (property)	`dict[str, Any]`

Members

reload_mcp

tai42_contract.app.facets.AppAdmin.reload_mcp

reload_mcp(self, title: str) -> dict[str, Any]

Parameters

Parameter	Type	Default	Description
`title`	`str`	—	—

deregister_mcp

tai42_contract.app.facets.AppAdmin.deregister_mcp

deregister_mcp(self, title: str) -> dict[str, Any]

Parameters

Parameter	Type	Default	Description
`title`	`str`	—	—

reload_config

tai42_contract.app.facets.AppAdmin.reload_config

reload_config(self) -> dict[str, Any]

tool_reloader

tai42_contract.app.facets.AppAdmin.tool_reloader

tool_reloader(self, kind: str) -> Callable[..., Any]

Parameter	Type	Default	Description
`kind`	`str`	—	—

run_tool_reload

tai42_contract.app.facets.AppAdmin.run_tool_reload

async run_tool_reload(self, kind: str, action: str, name: str) -> dict[str, Any]

Parameters

Parameter	Type	Default	Description
`kind`	`str`	—	—
`action`	`str`	—	—
`name`	`str`	—	—

reload_failed_mcps

tai42_contract.app.facets.AppAdmin.reload_failed_mcps

reload_failed_mcps(self) -> list[dict[str, Any]]

list_failed_mcps

tai42_contract.app.facets.AppAdmin.list_failed_mcps

list_failed_mcps(self) -> list[dict[str, Any]]

live_mcp_status

tai42_contract.app.facets.AppAdmin.live_mcp_status

live_mcp_status(self) -> dict[str, Any]

Snapshot the in-process MCP-binding state.

AppAgents

tai42_contract.app.facets.AppAgents

class AppAgents(Protocol)

Members

agent

tai42_contract.app.facets.AppAgents.agent

agent(self, name: str) -> Callable[[type[_AgentT]], type[_AgentT]]

Register an Agent subclass under name and auto-register its JSON run tool. The decorator returns the class unchanged, so the decorated symbol keeps its concrete subclass type. Parameters

Parameter	Type	Default	Description
`name`	`str`	—	—

get_agent

tai42_contract.app.facets.AppAgents.get_agent

get_agent(self, name: str) -> Agent

Fetch a registered agent instance by name; raise if missing. Parameters

Parameter	Type	Default	Description
`name`	`str`	—	—

all_agents

tai42_contract.app.facets.AppAgents.all_agents

all_agents(self) -> dict[str, Agent]

Every registered agent keyed by registration name (a shallow copy, so a caller iterating it cannot mutate the live registry). The preset bind kernel reads the keys to detect an agent base (the run tool binds under the registration name).

AppBackends

tai42_contract.app.facets.AppBackends

class AppBackends(Protocol)

Attributes

Attribute	Type
`backend` (property)	`Backend \| None`

Members

register_backend

tai42_contract.app.facets.AppBackends.register_backend

register_backend(self, cls: type[Backend] | None = None) -> Callable[[type[Backend]], type[Backend]] | type[Backend]

AppBackup

tai42_contract.app.facets.AppBackup

class AppBackup(Protocol)

Registry for named backup sections and the run of one section’s export/import. A plugin (or the host itself, the first consumer) registers a section under name by supplying an exporter() that returns a JSON-safe payload and an importer(payload) that applies it and returns a section report. sections() lists the registered sections for the UI. export_section / import_section run one section’s exporter/importer by name and raise loudly on an unknown name — never a silent no-op.

Members

register_section

tai42_contract.app.facets.AppBackup.register_section

register_section(
    self,
    name: str,
    exporter: Callable[[], Any],
    importer: Callable[[Any], Any],
    *,
    secret: bool = False,
) -> None

Parameters

Parameter	Type	Default	Description
`name`	`str`	—	—
`exporter`	`Callable[[], Any]`	—	—
`importer`	`Callable[[Any], Any]`	—	—
`secret`	`bool`	`False`	—

sections

tai42_contract.app.facets.AppBackup.sections

sections(self) -> list[BackupSectionInfo]

export_section

tai42_contract.app.facets.AppBackup.export_section

export_section(self, name: str) -> Any

Parameters

Parameter	Type	Default	Description
`name`	`str`	—	—

import_section

tai42_contract.app.facets.AppBackup.import_section

import_section(self, name: str, payload: Any) -> Any

Parameters

Parameter	Type	Default	Description
`name`	`str`	—	—
`payload`	`Any`	—	—

AppChannels

tai42_contract.app.facets.AppChannels

class AppChannels(Protocol)

Members

register

tai42_contract.app.facets.AppChannels.register

register(self, name: str, channel: Channel) -> None

Register a Channel under name. A channel plugin calls this through the tai42_app handle when its import-only channel_modules entry loads. Registering a name already taken raises loudly — a silent overwrite could swap the medium a live ask is delivered on. Parameters

Parameter	Type	Default	Description
`name`	`str`	—	—
`channel`	`Channel`	—	—

get

tai42_contract.app.facets.AppChannels.get

get(self, name: str) -> Channel

Fetch a registered channel by name; raise loudly on an unknown name. Resolution happens when ask_user is called with channel=name, BEFORE any interaction state is written, so an unknown name surfaces as a loud failure, never a question silently delivered nowhere. Parameters

Parameter	Type	Default	Description
`name`	`str`	—	—

names

tai42_contract.app.facets.AppChannels.names

names(self) -> list[str]

Every registered channel name, for the channels catalog route.

AppClients

tai42_contract.app.facets.AppClients

class AppClients(Protocol)

Members

client_ctx

tai42_contract.app.facets.AppClients.client_ctx

client_ctx(
    self,
    client_cls: type[BaseClient[ClientT]],
    settings: Any = None,
    *,
    fresh: bool = False,
    **kwargs: Any,
) -> AbstractAsyncContextManager[ClientT]

Async context manager yielding a connected client, pooled per loop + connection params (or one-shot when fresh=True). settings is a kit ClientSettings whose client_kwargs() supplies the connection key. Parameters

Parameter	Type	Default	Description
`client_cls`	`type[BaseClient[ClientT]]`	—	—
`settings`	`Any`	`None`	—
`fresh`	`bool`	`False`	—
`kwargs`	`Any`	`{}`	—

shutdown_clients

tai42_contract.app.facets.AppClients.shutdown_clients

async shutdown_clients(self) -> None

Close every live client pool for the running loop.

AppConfig

tai42_contract.app.facets.AppConfig

class AppConfig(Protocol)

Attributes

Attribute	Type
`config_manager` (property)	`ConfigManager`

AppConnectors

tai42_contract.app.facets.AppConnectors

class AppConnectors(Protocol)

Members

register_connector

tai42_contract.app.facets.AppConnectors.register_connector

register_connector(self, descriptor: ProviderDescriptor) -> None

Register an OAuth connector provider from its pure descriptor data. A provider plugin calls this through the tai42_app handle when the manifest loads it (a connector is pure data, so this is a plain call, not a decorator). Parameters

Parameter	Type	Default	Description
`descriptor`	`ProviderDescriptor`	—	—

token_store

tai42_contract.app.facets.AppConnectors.token_store

token_store: ConnectorTokenStore

The connector token store (single-namespace, keyed by connection_id).

AppExtensions

tai42_contract.app.facets.AppExtensions

class AppExtensions(Protocol)

Members

extension

tai42_contract.app.facets.AppExtensions.extension

extension(
    self,
    f: Callable[..., Any] | None = None,
    *,
    kind: ExtensionKind,
    name: str | None = None,
    requires_body_locality: bool = False,
) -> Callable[..., Any]

Register a tool-extension factory under name (default: the function’s own name); usable bare or with arguments. requires_body_locality marks an extension whose wrapper only works in the process running the tool body it wraps — e.g. a proxy layer that routes the body’s egress through a task-scoped contextvar, visible only where the body executes. The flag is stored as registration metadata the apply site reads to order a stacked combo: a locality-requiring extension must bind INSIDE any execution-relocating extension (ExtensionKind.relocates_execution), so its wrapper travels with the body to the worker. Bound outside a relocating layer, the wrapper stays behind in the submitting process and silently does not apply. Parameters

Parameter	Type	Default	Description
`f`	`Callable[..., Any] \| None`	`None`	—
`kind`	`ExtensionKind`	—	—
`name`	`str \| None`	`None`	—
`requires_body_locality`	`bool`	`False`	—

available_extensions

tai42_contract.app.facets.AppExtensions.available_extensions

available_extensions(self) -> list[dict[str, str]]

List registered extension modules as [{"name", "kind"}].

AppHttp

tai42_contract.app.facets.AppHttp

class AppHttp(Protocol)

Members

middleware

tai42_contract.app.facets.AppHttp.middleware

middleware(self, cls: type[Any] | None = None, **options: Any) -> Callable[..., Any]

Parameters

Parameter	Type	Default	Description
`options`	`Any`	`{}`	—

custom_route

tai42_contract.app.facets.AppHttp.custom_route

custom_route(
    self,
    path: str,
    methods: list[str],
    name: str | None = None,
    include_in_schema: bool = True,
    *,
    summary: str,
    tags: list[str],
    response_model: type[BaseModel] | None,
    request_model: type[BaseModel] | None = None,
    authed: bool = True,
) -> Callable[[Callable[[Request], Awaitable[Response]]], Callable[[Request], Awaitable[Response]]]

Register a Starlette handler AND its self-describing route metadata. The metadata is the single source of truth the OpenAPI 3.1 emitter and its coverage gate consume, so every registration MUST describe itself:

summary — a one-line operation summary (required, non-empty).
tags — at least one OpenAPI tag grouping the route (required).
response_model — the pydantic model wrapped in the {"data": ...} success envelope, or None as the explicit opaque marker for a route with no structured response. Required — there is no default, so a route cannot silently omit it.
request_model — the pydantic model of the request body; required for any route that reads a body, omitted (None) otherwise.
authed — whether the route requires the api key (default True); emitted as the security requirement.

The handler’s narrative docstring becomes the operation description, and the reload-gate 503 response is derived from the handler body. Parameters

Parameter	Type	Default	Description
`path`	`str`	—	—
`methods`	`list[str]`	—	—
`name`	`str \| None`	`None`	—
`include_in_schema`	`bool`	`True`	—
`summary`	`str`	—	—
`tags`	`list[str]`	—	—
`response_model`	`type[BaseModel] \| None`	—	—
`request_model`	`type[BaseModel] \| None`	`None`	—
`authed`	`bool`	`True`	—

AppLifecycle

tai42_contract.app.facets.AppLifecycle

class AppLifecycle(Protocol)

Members

on_startup

tai42_contract.app.facets.AppLifecycle.on_startup

on_startup(self, func: Callable[[], Any]) -> Callable[[], Any]

Parameters

Parameter	Type	Default	Description
`func`	`Callable[[], Any]`	—	—

on_shutdown

tai42_contract.app.facets.AppLifecycle.on_shutdown

on_shutdown(self, func: Callable[[], Any]) -> Callable[[], Any]

Parameters

Parameter	Type	Default	Description
`func`	`Callable[[], Any]`	—	—

on_reload

tai42_contract.app.facets.AppLifecycle.on_reload

on_reload(self, func: Callable[[], Any]) -> Callable[[], Any]

Register a handler re-run after every in-place re-init (reload_config) — e.g. dynamic tool loaders that on_startup ran once. Parameters

Parameter	Type	Default	Description
`func`	`Callable[[], Any]`	—	—

on_fleet_op_applied

tai42_contract.app.facets.AppLifecycle.on_fleet_op_applied

on_fleet_op_applied(self, func: Callable[[str], Any]) -> Callable[[str], Any]

Register a handler fired after any worker-bus op applies in this process, and after the reconnect self-resync reload. Unlike its zero-arg siblings, the handler takes ONE argument — the op name — so it can act on some ops and skip others (a query op carries no state change to react to). Parameters

Parameter	Type	Default	Description
`func`	`Callable[[str], Any]`	—	—

wait_until_ready

tai42_contract.app.facets.AppLifecycle.wait_until_ready

async wait_until_ready(self) -> None

Block until this process’s first boot self-resync has completed — the point at which the tool registry is fully built and stable for the run. A backend runtime that forks a child (or otherwise consumes queued work) per job awaits this before its work loop accepts anything: the boot self-resync rebuilds the tool registry non-atomically, so a worker that dequeued and forked mid-rebuild would run the job against a half-built registry. The latch is one-way — it resolves once and stays resolved, so a later bus reconnect (which re-runs the self-resync while the app is already live) never un-readies the process.

AppMonitoring

tai42_contract.app.facets.AppMonitoring

class AppMonitoring(Protocol)

Members

register_monitoring

tai42_contract.app.facets.AppMonitoring.register_monitoring

register_monitoring(self, builder: Callable[..., Any] | None = None) -> Callable[..., Any]

Parameters

Parameter	Type	Default	Description
`builder`	`Callable[..., Any] \| None`	`None`	—

active

tai42_contract.app.facets.AppMonitoring.active

active: Monitoring

The active monitoring backend (impl type behind the contract protocol).

AppPresets

tai42_contract.app.facets.AppPresets

class AppPresets(Protocol)

The presets namespace (app.presets) — the typed view + the bind kernel. store is the PresetStore-typed view over app.versioning.store (kind="preset"). bind is the kernel BOTH tiers (ephemeral and versioned) build their live tool through, so its typed-schema behavior reaches both. Attributes

Attribute	Type
`store` (property)	`PresetStore`

Members

bind

tai42_contract.app.facets.AppPresets.bind

async bind(
    self,
    base_tool: str,
    fixed_kwargs: dict[str, Any],
    *,
    name: str,
    description: str = '',
    tags: list[str] | None = None,
    output_schema: dict[str, Any] | None = None,
) -> Tool

Return a FastMCP tool transform of base_tool as a new named tool. Built in ONE Tool.from_tool call: each fixed_kwargs key is baked as a HIDDEN, FIXED constant (removed from the exposed schema; a caller that passes it is rejected — it cannot be overridden at runtime) while the REMAINING arguments keep the base tool’s real typed schema (names, types, descriptions). tags sets the transformed tool’s native tags. An output_schema (an object JSON Schema) is baked into an agent base’s response_format (forcing structured output) or advertised + validated on a plain tool base. bind is async because it must resolve the base Tool object (via app.tools.get_tool(base_tool)) to feed the transform. Parameters

Parameter	Type	Default	Description
`base_tool`	`str`	—	—
`fixed_kwargs`	`dict[str, Any]`	—	—
`name`	`str`	—	—
`description`	`str`	`''`	—
`tags`	`list[str] \| None`	`None`	—
`output_schema`	`dict[str, Any] \| None`	`None`	—

AppStorage

tai42_contract.app.facets.AppStorage

class AppStorage(Protocol)

Members

register_storage

tai42_contract.app.facets.AppStorage.register_storage

register_storage(self, cls: type[Storage] | None = None) -> Callable[[type[Storage]], type[Storage]] | type[Storage]

resource_manager

tai42_contract.app.facets.AppStorage.resource_manager

resource_manager: Any

The active resource manager (impl type; loads/renders content over storage).

AppSubApp

tai42_contract.app.facets.AppSubApp

class AppSubApp(Protocol)

Attributes

Attribute	Type
`mcp_sub_app_router` (property)	`SubMcpAppRouter`

AppTools

tai42_contract.tools.AppTools

class AppTools(Protocol)

Tool + toolkit registration / lookup surface.

Members

tool_title

tai42_contract.tools.AppTools.tool_title

tool_title(self, func: Callable[..., object]) -> str

Parameters

Parameter	Type	Default	Description
`func`	`Callable[..., object]`	—	—

get_tool

tai42_contract.tools.AppTools.get_tool

async get_tool(self, key: str) -> Tool

Parameters

Parameter	Type	Default	Description
`key`	`str`	—	—

get_tools

tai42_contract.tools.AppTools.get_tools

async get_tools(self) -> dict[str, Tool]

get_client_tools

tai42_contract.tools.AppTools.get_client_tools

async get_client_tools(self, names: list[str] | None = None) -> list[StructuredTool]

Parameters

Parameter	Type	Default	Description
`names`	`list[str] \| None`	`None`	—

run_tool

tai42_contract.tools.AppTools.run_tool

async run_tool(self, key: str, arguments: dict[str, Any], *, offload_sync: bool = False) -> Any

Parameters

Parameter	Type	Default	Description
`key`	`str`	—	—
`arguments`	`dict[str, Any]`	—	—
`offload_sync`	`bool`	`False`	—

remove_tool

tai42_contract.tools.AppTools.remove_tool

remove_tool(self, name: str) -> None

Parameters

Parameter	Type	Default	Description
`name`	`str`	—	—

register_tool_info

tai42_contract.tools.AppTools.register_tool_info

register_tool_info(self, name: str, combos: Sequence[Sequence[ExtensionElement]] | None = None) -> None

Parameters

Parameter	Type	Default	Description
`name`	`str`	—	—
`combos`	`Sequence[Sequence[ExtensionElement]] \| None`	`None`	—

unregister_tool_info

tai42_contract.tools.AppTools.unregister_tool_info

unregister_tool_info(self, name: str) -> None

Parameters

Parameter	Type	Default	Description
`name`	`str`	—	—

unregister_tool_base

tai42_contract.tools.AppTools.unregister_tool_base

unregister_tool_base(self, tool_name: str) -> list[str]

Parameters

Parameter	Type	Default	Description
`tool_name`	`str`	—	—

AppVersioning

tai42_contract.app.facets.AppVersioning

class AppVersioning(Protocol)

The generic versioned-document store namespace (app.versioning). This is the platform persistence primitive — append-only versions + an active pointer + rollback over an opaque JSONB body, discriminated by kind. Direct consumers (e.g. AC policies under kind="ac_policy") reach it here; presets reach it through AppPresets, never as a new kind. Attributes

Attribute	Type
`store` (property)	`VersionedStore`

AppWebhookVerifiers

tai42_contract.app.facets.AppWebhookVerifiers

class AppWebhookVerifiers(Protocol)

Members

register

tai42_contract.app.facets.AppWebhookVerifiers.register

register(self, name: str, verifier: WebhookVerifier) -> None

Register a WebhookVerifier under name. A provider plugin calls this through the tai42_app handle when its import-only webhook_verifier_modules entry loads. Registering a name already taken raises loudly — a silent overwrite could swap a topic’s verifier out from under a live binding. Parameters

Parameter	Type	Default	Description
`name`	`str`	—	—
`verifier`	`WebhookVerifier`	—	—

get

tai42_contract.app.facets.AppWebhookVerifiers.get

get(self, name: str) -> WebhookVerifier

Fetch a registered verifier by name; raise loudly on an unknown name. Resolution happens when a verifier is bound to a public webhook door, so an unknown name surfaces at bind time as a loud failure, never a silently-unverified door. Parameters

Parameter	Type	Default	Description
`name`	`str`	—	—

TaiApp

tai42_contract.app.TaiApp

class TaiApp(Protocol)

The assembled facade — per-feature sub-protocols exposed as namespaces. Attributes

Attribute	Type
`tools` (property)	`AppTools`
`agents` (property)	`AppAgents`
`backends` (property)	`AppBackends`
`storage` (property)	`AppStorage`
`connectors` (property)	`AppConnectors`
`webhook_verifiers` (property)	`AppWebhookVerifiers`
`channels` (property)	`AppChannels`
`monitoring` (property)	`AppMonitoring`
`extensions` (property)	`AppExtensions`
`http` (property)	`AppHttp`
`clients` (property)	`AppClients`
`lifecycle` (property)	`AppLifecycle`
`admin` (property)	`AppAdmin`
`config` (property)	`AppConfig`
`backup` (property)	`AppBackup`
`sub_app` (property)	`AppSubApp`
`versioning` (property)	`AppVersioning`
`presets` (property)	`AppPresets`

tai42_app

tai42_contract.app.tai42_app

tai42_app: _TaiAppRuntime = cast('_TaiAppRuntime', _tai_app_handle)

AppAgents

tai42_contract.app.facets.AppAgents

class AppAgents(Protocol)

Members

agent

tai42_contract.app.facets.AppAgents.agent

agent(self, name: str) -> Callable[[type[_AgentT]], type[_AgentT]]

Register an Agent subclass under name and auto-register its JSON run tool. The decorator returns the class unchanged, so the decorated symbol keeps its concrete subclass type. Parameters

Parameter	Type	Default	Description
`name`	`str`	—	—

get_agent

tai42_contract.app.facets.AppAgents.get_agent

get_agent(self, name: str) -> Agent

Fetch a registered agent instance by name; raise if missing. Parameters

Parameter	Type	Default	Description
`name`	`str`	—	—

all_agents

tai42_contract.app.facets.AppAgents.all_agents

all_agents(self) -> dict[str, Agent]

AppBackends

tai42_contract.app.facets.AppBackends

class AppBackends(Protocol)

Attributes

Attribute	Type
`backend` (property)	`Backend \| None`

Members

register_backend

tai42_contract.app.facets.AppBackends.register_backend

register_backend(self, cls: type[Backend] | None = None) -> Callable[[type[Backend]], type[Backend]] | type[Backend]

AppStorage

tai42_contract.app.facets.AppStorage

class AppStorage(Protocol)

Members

register_storage

tai42_contract.app.facets.AppStorage.register_storage

register_storage(self, cls: type[Storage] | None = None) -> Callable[[type[Storage]], type[Storage]] | type[Storage]

resource_manager

tai42_contract.app.facets.AppStorage.resource_manager

resource_manager: Any

The active resource manager (impl type; loads/renders content over storage).

AppMonitoring

tai42_contract.app.facets.AppMonitoring

class AppMonitoring(Protocol)

Members

register_monitoring

tai42_contract.app.facets.AppMonitoring.register_monitoring

register_monitoring(self, builder: Callable[..., Any] | None = None) -> Callable[..., Any]

Parameters

Parameter	Type	Default	Description
`builder`	`Callable[..., Any] \| None`	`None`	—

active

tai42_contract.app.facets.AppMonitoring.active

active: Monitoring

The active monitoring backend (impl type behind the contract protocol).

AppExtensions

tai42_contract.app.facets.AppExtensions

class AppExtensions(Protocol)

Members

extension

tai42_contract.app.facets.AppExtensions.extension

extension(
    self,
    f: Callable[..., Any] | None = None,
    *,
    kind: ExtensionKind,
    name: str | None = None,
    requires_body_locality: bool = False,
) -> Callable[..., Any]

Parameter	Type	Default	Description
`f`	`Callable[..., Any] \| None`	`None`	—
`kind`	`ExtensionKind`	—	—
`name`	`str \| None`	`None`	—
`requires_body_locality`	`bool`	`False`	—

available_extensions

tai42_contract.app.facets.AppExtensions.available_extensions

available_extensions(self) -> list[dict[str, str]]

List registered extension modules as [{"name", "kind"}].

AppWebhookVerifiers

tai42_contract.app.facets.AppWebhookVerifiers

class AppWebhookVerifiers(Protocol)

Members

register

tai42_contract.app.facets.AppWebhookVerifiers.register

register(self, name: str, verifier: WebhookVerifier) -> None

Parameter	Type	Default	Description
`name`	`str`	—	—
`verifier`	`WebhookVerifier`	—	—

get

tai42_contract.app.facets.AppWebhookVerifiers.get

get(self, name: str) -> WebhookVerifier

Parameter	Type	Default	Description
`name`	`str`	—	—

AppChannels

tai42_contract.app.facets.AppChannels

class AppChannels(Protocol)

Members

register

tai42_contract.app.facets.AppChannels.register

register(self, name: str, channel: Channel) -> None

Parameter	Type	Default	Description
`name`	`str`	—	—
`channel`	`Channel`	—	—

get

tai42_contract.app.facets.AppChannels.get

get(self, name: str) -> Channel

Parameter	Type	Default	Description
`name`	`str`	—	—

names

tai42_contract.app.facets.AppChannels.names

names(self) -> list[str]

Every registered channel name, for the channels catalog route.

AppConnectors

tai42_contract.app.facets.AppConnectors

class AppConnectors(Protocol)

Members

register_connector

tai42_contract.app.facets.AppConnectors.register_connector

register_connector(self, descriptor: ProviderDescriptor) -> None

Parameter	Type	Default	Description
`descriptor`	`ProviderDescriptor`	—	—

token_store

tai42_contract.app.facets.AppConnectors.token_store

token_store: ConnectorTokenStore

The connector token store (single-namespace, keyed by connection_id).

AppHttp

tai42_contract.app.facets.AppHttp

class AppHttp(Protocol)

Members

middleware

tai42_contract.app.facets.AppHttp.middleware

middleware(self, cls: type[Any] | None = None, **options: Any) -> Callable[..., Any]

Parameters

Parameter	Type	Default	Description
`options`	`Any`	`{}`	—

custom_route

tai42_contract.app.facets.AppHttp.custom_route

custom_route(
    self,
    path: str,
    methods: list[str],
    name: str | None = None,
    include_in_schema: bool = True,
    *,
    summary: str,
    tags: list[str],
    response_model: type[BaseModel] | None,
    request_model: type[BaseModel] | None = None,
    authed: bool = True,
) -> Callable[[Callable[[Request], Awaitable[Response]]], Callable[[Request], Awaitable[Response]]]

summary — a one-line operation summary (required, non-empty).
tags — at least one OpenAPI tag grouping the route (required).
response_model — the pydantic model wrapped in the {"data": ...} success envelope, or None as the explicit opaque marker for a route with no structured response. Required — there is no default, so a route cannot silently omit it.
request_model — the pydantic model of the request body; required for any route that reads a body, omitted (None) otherwise.
authed — whether the route requires the api key (default True); emitted as the security requirement.

The handler’s narrative docstring becomes the operation description, and the reload-gate 503 response is derived from the handler body. Parameters

Parameter	Type	Default	Description
`path`	`str`	—	—
`methods`	`list[str]`	—	—
`name`	`str \| None`	`None`	—
`include_in_schema`	`bool`	`True`	—
`summary`	`str`	—	—
`tags`	`list[str]`	—	—
`response_model`	`type[BaseModel] \| None`	—	—
`request_model`	`type[BaseModel] \| None`	`None`	—
`authed`	`bool`	`True`	—

AppClients

tai42_contract.app.facets.AppClients

class AppClients(Protocol)

Members

client_ctx

tai42_contract.app.facets.AppClients.client_ctx

client_ctx(
    self,
    client_cls: type[BaseClient[ClientT]],
    settings: Any = None,
    *,
    fresh: bool = False,
    **kwargs: Any,
) -> AbstractAsyncContextManager[ClientT]

Parameter	Type	Default	Description
`client_cls`	`type[BaseClient[ClientT]]`	—	—
`settings`	`Any`	`None`	—
`fresh`	`bool`	`False`	—
`kwargs`	`Any`	`{}`	—

shutdown_clients

tai42_contract.app.facets.AppClients.shutdown_clients

async shutdown_clients(self) -> None

Close every live client pool for the running loop.

AppLifecycle

tai42_contract.app.facets.AppLifecycle

class AppLifecycle(Protocol)

Members

on_startup

tai42_contract.app.facets.AppLifecycle.on_startup

on_startup(self, func: Callable[[], Any]) -> Callable[[], Any]

Parameters

Parameter	Type	Default	Description
`func`	`Callable[[], Any]`	—	—

on_shutdown

tai42_contract.app.facets.AppLifecycle.on_shutdown

on_shutdown(self, func: Callable[[], Any]) -> Callable[[], Any]

Parameters

Parameter	Type	Default	Description
`func`	`Callable[[], Any]`	—	—

on_reload

tai42_contract.app.facets.AppLifecycle.on_reload

on_reload(self, func: Callable[[], Any]) -> Callable[[], Any]

Register a handler re-run after every in-place re-init (reload_config) — e.g. dynamic tool loaders that on_startup ran once. Parameters

Parameter	Type	Default	Description
`func`	`Callable[[], Any]`	—	—

on_fleet_op_applied

tai42_contract.app.facets.AppLifecycle.on_fleet_op_applied

on_fleet_op_applied(self, func: Callable[[str], Any]) -> Callable[[str], Any]

Parameter	Type	Default	Description
`func`	`Callable[[str], Any]`	—	—

wait_until_ready

tai42_contract.app.facets.AppLifecycle.wait_until_ready

async wait_until_ready(self) -> None

AppAdmin

tai42_contract.app.facets.AppAdmin

class AppAdmin(Protocol)

Attributes

Attribute	Type
`live_manifest` (property)	`dict[str, Any]`

Members

reload_mcp

tai42_contract.app.facets.AppAdmin.reload_mcp

reload_mcp(self, title: str) -> dict[str, Any]

Parameters

Parameter	Type	Default	Description
`title`	`str`	—	—

deregister_mcp

tai42_contract.app.facets.AppAdmin.deregister_mcp

deregister_mcp(self, title: str) -> dict[str, Any]

Parameters

Parameter	Type	Default	Description
`title`	`str`	—	—

reload_config

tai42_contract.app.facets.AppAdmin.reload_config

reload_config(self) -> dict[str, Any]

tool_reloader

tai42_contract.app.facets.AppAdmin.tool_reloader

tool_reloader(self, kind: str) -> Callable[..., Any]

Parameter	Type	Default	Description
`kind`	`str`	—	—

run_tool_reload

tai42_contract.app.facets.AppAdmin.run_tool_reload

async run_tool_reload(self, kind: str, action: str, name: str) -> dict[str, Any]

Parameters

Parameter	Type	Default	Description
`kind`	`str`	—	—
`action`	`str`	—	—
`name`	`str`	—	—

reload_failed_mcps

tai42_contract.app.facets.AppAdmin.reload_failed_mcps

reload_failed_mcps(self) -> list[dict[str, Any]]

list_failed_mcps

tai42_contract.app.facets.AppAdmin.list_failed_mcps

list_failed_mcps(self) -> list[dict[str, Any]]

live_mcp_status

tai42_contract.app.facets.AppAdmin.live_mcp_status

live_mcp_status(self) -> dict[str, Any]

Snapshot the in-process MCP-binding state.

AppConfig

tai42_contract.app.facets.AppConfig

class AppConfig(Protocol)

Attributes

Attribute	Type
`config_manager` (property)	`ConfigManager`

AppBackup

tai42_contract.app.facets.AppBackup

class AppBackup(Protocol)

Members

register_section

tai42_contract.app.facets.AppBackup.register_section

register_section(
    self,
    name: str,
    exporter: Callable[[], Any],
    importer: Callable[[Any], Any],
    *,
    secret: bool = False,
) -> None

Parameters

Parameter	Type	Default	Description
`name`	`str`	—	—
`exporter`	`Callable[[], Any]`	—	—
`importer`	`Callable[[Any], Any]`	—	—
`secret`	`bool`	`False`	—

sections

tai42_contract.app.facets.AppBackup.sections

sections(self) -> list[BackupSectionInfo]

export_section

tai42_contract.app.facets.AppBackup.export_section

export_section(self, name: str) -> Any

Parameters

Parameter	Type	Default	Description
`name`	`str`	—	—

import_section

tai42_contract.app.facets.AppBackup.import_section

import_section(self, name: str, payload: Any) -> Any

Parameters

Parameter	Type	Default	Description
`name`	`str`	—	—
`payload`	`Any`	—	—

AppSubApp

tai42_contract.app.facets.AppSubApp

class AppSubApp(Protocol)

Attributes

Attribute	Type
`mcp_sub_app_router` (property)	`SubMcpAppRouter`

AppVersioning

tai42_contract.app.facets.AppVersioning

class AppVersioning(Protocol)

Attribute	Type
`store` (property)	`VersionedStore`

AppPresets

tai42_contract.app.facets.AppPresets

class AppPresets(Protocol)

Attribute	Type
`store` (property)	`PresetStore`

Members

bind

tai42_contract.app.facets.AppPresets.bind

async bind(
    self,
    base_tool: str,
    fixed_kwargs: dict[str, Any],
    *,
    name: str,
    description: str = '',
    tags: list[str] | None = None,
    output_schema: dict[str, Any] | None = None,
) -> Tool

Parameter	Type	Default	Description
`base_tool`	`str`	—	—
`fixed_kwargs`	`dict[str, Any]`	—	—
`name`	`str`	—	—
`description`	`str`	`''`	—
`tags`	`list[str] \| None`	`None`	—
`output_schema`	`dict[str, Any] \| None`	`None`	—

Python SDK Agents (tai42_contract.agent)

⌘I

​AppAdmin

​Members

​reload_mcp

​deregister_mcp

​reload_config

​tool_reloader

​run_tool_reload

​reload_failed_mcps

​list_failed_mcps

​live_mcp_status

​AppAgents

​Members

​agent

​get_agent

​all_agents

​AppBackends

​Members

​register_backend

​AppBackup

​Members

​register_section

​sections

​export_section

​import_section

​AppChannels

​Members

​register

​get

​names

​AppClients

​Members

​client_ctx

​shutdown_clients

​AppConfig

​AppConnectors

​Members

​register_connector

​token_store

​AppExtensions

​Members

​extension

​available_extensions

​AppHttp

​Members

​middleware

​custom_route

​AppLifecycle

​Members

​on_startup

​on_shutdown

​on_reload

​on_fleet_op_applied

​wait_until_ready

​AppMonitoring

​Members

​register_monitoring

​active

​AppPresets

​Members

​bind

​AppStorage

​Members

​register_storage

​resource_manager

​AppSubApp

​AppTools

​Members

​tool_title

​get_tool

​get_tools

​get_client_tools

​run_tool

​remove_tool

​register_tool_info

​unregister_tool_info

​unregister_tool_base

​AppVersioning

​AppWebhookVerifiers

​Members

​register

AppAdmin

Members

reload_mcp

deregister_mcp

reload_config

tool_reloader

run_tool_reload

reload_failed_mcps

list_failed_mcps

live_mcp_status

AppAgents

Members

agent

get_agent

all_agents

AppBackends

Members

register_backend

AppBackup

Members

register_section

sections

export_section

import_section

AppChannels

Members

register

get

names

AppClients

Members

client_ctx

shutdown_clients

AppConfig

AppConnectors

Members

register_connector

token_store

AppExtensions

Members

extension

available_extensions

AppHttp

Members

middleware

custom_route

AppLifecycle

Members

on_startup

on_shutdown

on_reload

on_fleet_op_applied

wait_until_ready

AppMonitoring

Members

register_monitoring

active

AppPresets

Members

bind

AppStorage

Members

register_storage

resource_manager

AppSubApp

AppTools

Members

tool_title

get_tool

get_tools

get_client_tools

run_tool

remove_tool

register_tool_info

unregister_tool_info

unregister_tool_base

AppVersioning

AppWebhookVerifiers

Members

register