API Reference¶

Public entrypoints¶

Taimoe Platform SDK for managed agent runtimes.

exception taimoe.platform.TaimoeAPIError(message, *, status_code=0, code=None, request_id=None, details=None, response=None)[source]

Bases: TaimoeError

Raised when an API request to the Taimoe Platform fails.

Parameters:

message (str)
status_code (int)
code (str | None)
request_id (str | None)
details (dict[str, Any] | None)
response (httpx.Response | None)

Return type:

None

status_code: HTTP status from the response (or 0 for transport-level).

code: Canonical logical code (e.g. "RESOURCE_EXHAUSTED") if the server returned a structured detail, else None.

message: Human-readable message extracted from the response.

request_id: X-Request-ID echoed by the server, for log correlation.

details: Any additional fields from the structured detail body.

response: The raw httpx Response, kept for advanced debugging.

default_code: ClassVar[str | None] = None: Default canonical code for the subclass. Overridden by concrete classes.

classmethod from_response(response)[source]

Build the most specific subclass for an HTTP error response.

Resolution order:

structured detail.code → _CODE_REGISTRY
HTTP status → _STATUS_REGISTRY
fall back to TaimoeAPIError

Parameters:: response (httpx.Response)
Return type:: TaimoeAPIError

class taimoe.platform.TaimoeClient(*, base_url, api_key=None, org='default', timeout=10.0, max_retries=2)[source]

Bases: object

REST client for the Taimoe Platform.

Follows Google ADK style: one client class; async operations use the *_async suffix on the same class instead of a separate AsyncTaimoeClient.

The client owns a long-lived httpx.Client and httpx.AsyncClient so connection pools and keep-alive are reused across requests. Use as a context manager or call close() / aclose() when done.

Retries are opt-in via max_retries (default 2). The client retries network errors and HTTP 408/429/5xx with exponential backoff, honouring the server’s Retry-After header when present. Set max_retries=0 to disable retries entirely.

Every request carries an X-Request-ID header for audit correlation. A UUID4 is generated automatically; callers can override per request by passing request_id=... through the resource layer.

Parameters:

base_url (str)
api_key (str | None)
org (str)
timeout (float | httpx.Timeout)
max_retries (int)

close()[source]

Close the underlying synchronous HTTP client.

Return type:: None

async aclose()[source]

Close the underlying asynchronous HTTP client.

Return type:: None

exception taimoe.platform.TaimoeError[source]

Bases: Exception

Base for every error raised by the Taimoe Platform SDK.

class taimoe.platform.TaimoeRegistry(*, runtime_name, platform_url, platform_token=None, runtime_id=None, poll_interval_seconds=300, framework=None, framework_version=None, on_sync_error=None)[source]

Bases: object

Coordinates discovery, sync, and local config access for a runtime service.

Construction does not touch global SDK state. To wire this registry into the global taimoe handle (so user code can call taimoe.agent(name) and resolve against this registry’s cache), do it explicitly:

from taimoe.platform.adapters.adk import taimoe
registry = TaimoeRegistry(...)
taimoe.bind_to(registry)

This keeps tests and multi-tenant setups cleanly separable.

Parameters:

runtime_name (str | None)
platform_url (str)
platform_token (str | None)
runtime_id (str | None)
poll_interval_seconds (float)
framework (str | None)
framework_version (str | None)
on_sync_error (SyncErrorHandler | None)

register(agent)[source]

Parameters:: agent (Any)
Return type:: None

register_many(agents)[source]

Parameters:: agents (Iterable[Any])
Return type:: None

property registered_agents: tuple[Any, ...]

runtime_manifest()[source]

Return type:: RuntimeManifest

agents_manifest()[source]

Return type:: AgentsManifest

health()[source]

Return type:: HealthStatus

sync_once(*, ignore_errors=True)[source]

Run one synchronous sync pass on both syncers.

ignore_errors defaults to True so a single call site can do cold-start init without each branch having to choose: failure logs a warning, the SDK enters degraded mode, and the next background tick retries. Pass ignore_errors=False if a caller needs the exception to bubble up (e.g. CI / tests).

Parameters:: ignore_errors (bool)
Return type:: None

start_sync()[source]

Return type:: None

stop_sync(timeout=None)[source]

Parameters:: timeout (float | None)
Return type:: None

fastapi_router()[source]

Return a single combined FastAPI router for discovery endpoints.

Convenient but must be mounted at the FastAPI app root to keep the well-known URI RFC 8615-compliant. Prefer install_routes() for new code — it wires both routers with the correct mounts.

Return type:: Any

install_routes(app)[source]

Wire well-known and discovery routes into a FastAPI app.

Equivalent to calling taimoe.platform.routes.install_routes() with this registry, just spelled as a method for ergonomics.

Parameters:: app (Any)
Return type:: None

class taimoe.platform.TaimoeSpan(*, trace, span_type, name, parent_span_id=None, input=None, attributes=None, agent_id=None)[source]

Bases: object

Context manager for one agent, LLM, tool, app, or policy operation.

Parameters:

trace (TaimoeTrace)
span_type (SpanType)
name (str)
parent_span_id (str | None)
input (Any)
attributes (dict[str, Any] | None)
agent_id (str | None)

set_input(value)[source]

Parameters:: value (Any)
Return type:: None

set_output(value)[source]

Parameters:: value (Any)
Return type:: None

set_attribute(key, value)[source]

Parameters:

key (str)
value (Any)

Return type:

None

set_usage(*, prompt_tokens=None, completion_tokens=None, cost_usd=None)[source]

Parameters:

prompt_tokens (int | None)
completion_tokens (int | None)
cost_usd (float | None)

Return type:

None

set_error(error)[source]

Parameters:: error (BaseException | str)
Return type:: None

finish()[source]

Return type:: SpanEvent

class taimoe.platform.TaimoeTrace(name, *, trace_id=None, session_id=None, agent_id=None, team_id=None, attributes=None, state=<taimoe.platform.observability.trace.ObservabilityState object>)[source]

Bases: object

Context manager representing one agent run or conversation trace.

Parameters:

name (str)
trace_id (str | None)
session_id (str | None)
agent_id (str | None)
team_id (str | None)
attributes (dict[str, Any] | None)
state (ObservabilityState)

property spans: tuple[SpanEvent, ...]

span(span_type, *, name, input=None, attributes=None, agent_id=None)[source]

Parameters:

span_type (Literal['agent', 'llm', 'tool', 'policy', 'workflow', 'app', 'custom'])
name (str)
input (Any)
attributes (dict[str, Any] | None)
agent_id (str | None)

Return type:

TaimoeSpan

record_span(span)[source]

Parameters:: span (SpanEvent)
Return type:: None

flush()[source]

Return type:: None

taimoe.platform.current_span()[source]

Return the active span for the current context.

Return type:: TaimoeSpan | None

taimoe.platform.current_trace()[source]

Return the active trace for the current context.

Return type:: TaimoeTrace | None

taimoe.platform.get_registry()[source]

Return the global registry instance, or raise if not initialized.

Return type:: TaimoeRegistry

taimoe.platform.init(api_key=None, runtime_name=None, platform_url=None, runtime_id=None, framework=None, framework_version=None, poll_interval_seconds=300, on_sync_error=None, instrument=None)[source]

Initialize the Taimoe Platform SDK.

Two operating modes (see TaimoeRegistry for details):

Runtime-bound — pass runtime_name. Pulls every agent under that runtime via one batched HTTP call.
Runtime-less — omit runtime_name. Pulls individual agents on demand whenever user code does taimoe.agent("foo").

Both modes can run together: with runtime_name set, on-demand taimoe.agent(name) calls still get individually polled.

Parameters:

api_key (str | None) – Gateway API key. Falls back to TAIMOE_VIRTUAL_KEY then (with a deprecation warning) TAIMOE_RUNTIME_TOKEN.
runtime_name (str | None) – Runtime identifier; enables batched sync. Falls back to TAIMOE_RUNTIME_NAME.
platform_url (str | None) – Gateway base URL. Falls back to TAIMOE_PLATFORM_URL and finally to http://localhost:8000 (development only).
runtime_id (str | None) – Override for the runtime identifier used in observability spans. Defaults to runtime_name when omitted.
framework (str | None) – Framework name advertised in the discovery manifest (e.g. "google-adk").
framework_version (str | None) – Framework version advertised in the manifest.
poll_interval_seconds (float) – How often the background syncers refresh.
on_sync_error (Callable[[Exception], None] | None) – Callback invoked when a background sync iteration fails after exhausting its in-loop retries.
instrument (bool | None) – Whether to apply auto-instrumentation (currently Google ADK). When None (default), the SDK auto-detects: instrument iff google.adk is already imported in this process. Pass True / False to override.

Return type:

TaimoeRegistry

Idempotent: calling init() twice in the same process returns the existing registry and logs a warning. Tests should call _reset_for_tests() to clear state between runs.

taimoe.platform.trace(name, *, trace_id=None, session_id=None, agent_id=None, team_id=None, attributes=None)[source]

Start a trace for one agent run or conversation.

Parameters:

name (str)
trace_id (str | None)
session_id (str | None)
agent_id (str | None)
team_id (str | None)
attributes (dict[str, Any] | None)

Return type:

TaimoeTrace

taimoe.platform.track_action(name=None)[source]

Wrap a function as a tool span on the active trace.

If there is no active trace the call passes through unwrapped (we don’t silently mint a trace for a single action — that would obscure which agent owns the action). Works for sync and async def callables.

Parameters:: name (str | None)
Return type:: Callable[[F], F]

taimoe.platform.track_agent(name=None)[source]

Wrap an agent entry point as an agent span, minting a trace if needed.

Works for sync and async def callables.

Parameters:: name (str | None)
Return type:: Callable[[F], F]

Client¶

Synchronous and asynchronous Taimoe Platform client.

class taimoe.platform.client.client.TaimoeClient(*, base_url, api_key=None, org='default', timeout=10.0, max_retries=2)[source]¶

Bases: object

REST client for the Taimoe Platform.

Follows Google ADK style: one client class; async operations use the *_async suffix on the same class instead of a separate AsyncTaimoeClient.

The client owns a long-lived httpx.Client and httpx.AsyncClient so connection pools and keep-alive are reused across requests. Use as a context manager or call close() / aclose() when done.

Every request carries an X-Request-ID header for audit correlation. A UUID4 is generated automatically; callers can override per request by passing request_id=... through the resource layer.

Parameters:

base_url (str)
api_key (str | None)
org (str)
timeout (float | httpx.Timeout)
max_retries (int)

close()[source]¶

Close the underlying synchronous HTTP client.

Return type:: None

async aclose()[source]¶

Close the underlying asynchronous HTTP client.

Return type:: None

Registry and initialization¶

Top-level taimoe.init() entry point.

Environment variables consumed when the matching argument is omitted:

`TAIMOE_VIRTUAL_KEY`	API key issued by the Taimoe Gateway.
`TAIMOE_RUNTIME_TOKEN`	Deprecated alias for `TAIMOE_VIRTUAL_KEY` — emits a `DeprecationWarning` and will be removed in a future minor release.
`TAIMOE_PLATFORM_URL`	Base URL for the Taimoe Gateway. Defaults to `http://localhost:8000` for local development; production callers must pass this explicitly — the default is a footgun on hosts where no such service exists locally and the failure modes are silent (degraded sync, dropped audit).
`TAIMOE_RUNTIME_NAME`	Runtime identifier for the runtime-bound sync path. Omit for runtime-less mode.

taimoe.platform.init.init(api_key=None, runtime_name=None, platform_url=None, runtime_id=None, framework=None, framework_version=None, poll_interval_seconds=300, on_sync_error=None, instrument=None)[source]¶

Initialize the Taimoe Platform SDK.

Two operating modes (see TaimoeRegistry for details):

Runtime-bound — pass runtime_name. Pulls every agent under that runtime via one batched HTTP call.
Runtime-less — omit runtime_name. Pulls individual agents on demand whenever user code does taimoe.agent("foo").

Both modes can run together: with runtime_name set, on-demand taimoe.agent(name) calls still get individually polled.

Parameters:

api_key (str | None) – Gateway API key. Falls back to TAIMOE_VIRTUAL_KEY then (with a deprecation warning) TAIMOE_RUNTIME_TOKEN.
runtime_name (str | None) – Runtime identifier; enables batched sync. Falls back to TAIMOE_RUNTIME_NAME.
platform_url (str | None) – Gateway base URL. Falls back to TAIMOE_PLATFORM_URL and finally to http://localhost:8000 (development only).
runtime_id (str | None) – Override for the runtime identifier used in observability spans. Defaults to runtime_name when omitted.
framework (str | None) – Framework name advertised in the discovery manifest (e.g. "google-adk").
framework_version (str | None) – Framework version advertised in the manifest.
poll_interval_seconds (float) – How often the background syncers refresh.
on_sync_error (Callable[[Exception], None] | None) – Callback invoked when a background sync iteration fails after exhausting its in-loop retries.
instrument (bool | None) – Whether to apply auto-instrumentation (currently Google ADK). When None (default), the SDK auto-detects: instrument iff google.adk is already imported in this process. Pass True / False to override.

Return type:

TaimoeRegistry

Idempotent: calling init() twice in the same process returns the existing registry and logs a warning. Tests should call _reset_for_tests() to clear state between runs.

taimoe.platform.init.get_registry()[source]¶

Return the global registry instance, or raise if not initialized.

Return type:: TaimoeRegistry

Runtime registry exposed by customer services.

class taimoe.platform.registry.TaimoeRegistry(*, runtime_name, platform_url, platform_token=None, runtime_id=None, poll_interval_seconds=300, framework=None, framework_version=None, on_sync_error=None)[source]¶

Bases: object

Coordinates discovery, sync, and local config access for a runtime service.

from taimoe.platform.adapters.adk import taimoe
registry = TaimoeRegistry(...)
taimoe.bind_to(registry)

This keeps tests and multi-tenant setups cleanly separable.

Parameters:

runtime_name (str | None)
platform_url (str)
platform_token (str | None)
runtime_id (str | None)
poll_interval_seconds (float)
framework (str | None)
framework_version (str | None)
on_sync_error (SyncErrorHandler | None)

register(agent)[source]¶

Parameters:: agent (Any)
Return type:: None

register_many(agents)[source]¶

Parameters:: agents (Iterable[Any])
Return type:: None

property registered_agents: tuple[Any, ...]¶

runtime_manifest()[source]¶

Return type:: RuntimeManifest

agents_manifest()[source]¶

Return type:: AgentsManifest

health()[source]¶

Return type:: HealthStatus

sync_once(*, ignore_errors=True)[source]¶

Run one synchronous sync pass on both syncers.

Parameters:: ignore_errors (bool)
Return type:: None

start_sync()[source]¶

Return type:: None

stop_sync(timeout=None)[source]¶

Parameters:: timeout (float | None)
Return type:: None

fastapi_router()[source]¶

Return a single combined FastAPI router for discovery endpoints.

Convenient but must be mounted at the FastAPI app root to keep the well-known URI RFC 8615-compliant. Prefer install_routes() for new code — it wires both routers with the correct mounts.

Return type:: Any

install_routes(app)[source]¶

Wire well-known and discovery routes into a FastAPI app.

Equivalent to calling taimoe.platform.routes.install_routes() with this registry, just spelled as a method for ergonomics.

Parameters:: app (Any)
Return type:: None

ADK adapter¶

Google ADK adapter entrypoints.

class taimoe.platform.adapters.adk.TaimoeAgentHandle(runtime_agent_id, cache=None)[source]

Bases: object

Lazy handle that resolves the latest platform config for an ADK agent.

Use directly in LlmAgent(...):

agent = taimoe.agent("my_agent")
root_agent = LlmAgent(
    name="my_agent",
    model=agent.model,
    instruction=agent.instruction,
    tools=agent.tools,
)

Each attribute returns a value shaped for the ADK constructor field: instruction is a callable (InstructionProvider, ADK re-resolves per call), model and tools snapshot the current platform values at construction. The refresh_callback swaps agent.model / agent.tools in place on subsequent calls so platform edits take effect within one sync cycle.

instruction and model are intentionally asymmetric in their “no config yet” behavior:

instruction falls back to a bootstrap string — it’s user-visible and we want the agent to respond gracefully during cold start.
model raises TaimoeConfigUnavailableError — it’s a gateway-visible identifier and guessing would produce a confusing 404 downstream.

Parameters:

runtime_agent_id (str)
cache (RuntimeConfigCache | None)

bind_cache(cache)[source]

Parameters:: cache (RuntimeConfigCache)
Return type:: TaimoeAgentHandle

property config: AgentRuntimeConfig | None

property instruction: _InstructionProvider: ADK-compatible InstructionProvider. Resolved on every LLM call.

property model: str

Current model alias from the platform.

Raises TaimoeConfigUnavailableError if the platform config hasn’t been synced yet — see the class docstring for why this is louder than the instruction fallback.

property tools: list[Any]

Current tool list from the platform. Empty list when no config.

Note: the platform currently sends tool names/ids (list[str]), not real ADK tool objects. LlmAgent.tools expects list[BaseTool], so passing agent.tools straight in will fail at ADK construction. Resolve the names to your tool instances first, e.g.:

adk_tools = [my_tool_registry[name] for name in agent.tools]

Server-side tool-object resolution is a phase-2 feature.

property generation_config: dict[str, Any]

Current generation parameters (temperature, top_p, top_k, max_output_tokens, …) as defined in the console’s Model Settings.

Returns an empty dict when nothing has been configured — callers should treat absent keys as “use the model’s default”, not “set to 0”.

refresh_callback(*args, **kwargs)[source]

ADK before_agent_callback that swaps agent.model / agent.tools on every call. Wire this in addition to instruction=handle.instruction to get all three fields live-updating from the platform.

Parameters:

args (Any)
kwargs (Any)

Return type:

None

ADK-facing config handles.

exception taimoe.platform.adapters.adk.handle.TaimoeConfigUnavailableError[source]¶

Bases: RuntimeError

Raised when a handle is asked for a value the platform hasn’t synced.

We deliberately do not invent a fallback (e.g. a hardcoded model alias) — sending the wrong alias to the gateway would just produce a confusing 404 downstream. Failing loudly here points at the real cause: the agent isn’t registered on the platform under this runtime_agent_id, or the platform was unreachable when taimoe.init() ran.

class taimoe.platform.adapters.adk.handle.TaimoeAgentHandle(runtime_agent_id, cache=None)[source]¶

Bases: object

Lazy handle that resolves the latest platform config for an ADK agent.

Use directly in LlmAgent(...):

agent = taimoe.agent("my_agent")
root_agent = LlmAgent(
    name="my_agent",
    model=agent.model,
    instruction=agent.instruction,
    tools=agent.tools,
)

instruction and model are intentionally asymmetric in their “no config yet” behavior:

instruction falls back to a bootstrap string — it’s user-visible and we want the agent to respond gracefully during cold start.
model raises TaimoeConfigUnavailableError — it’s a gateway-visible identifier and guessing would produce a confusing 404 downstream.

Parameters:

runtime_agent_id (str)
cache (RuntimeConfigCache | None)

bind_cache(cache)[source]¶

Parameters:: cache (RuntimeConfigCache)
Return type:: TaimoeAgentHandle

property config: AgentRuntimeConfig | None¶

property instruction: _InstructionProvider¶: ADK-compatible InstructionProvider. Resolved on every LLM call.

property model: str¶

Current model alias from the platform.

Raises TaimoeConfigUnavailableError if the platform config hasn’t been synced yet — see the class docstring for why this is louder than the instruction fallback.

property tools: list[Any]¶

Current tool list from the platform. Empty list when no config.

adk_tools = [my_tool_registry[name] for name in agent.tools]

Server-side tool-object resolution is a phase-2 feature.

property generation_config: dict[str, Any]¶

Current generation parameters (temperature, top_p, top_k, max_output_tokens, …) as defined in the console’s Model Settings.

Returns an empty dict when nothing has been configured — callers should treat absent keys as “use the model’s default”, not “set to 0”.

refresh_callback(*args, **kwargs)[source]¶

ADK before_agent_callback that swaps agent.model / agent.tools on every call. Wire this in addition to instruction=handle.instruction to get all three fields live-updating from the platform.

Parameters:

args (Any)
kwargs (Any)

Return type:

None

class taimoe.platform.adapters.adk.handle.TaimoePromptHandle(runtime_prompt_id, cache=None)[source]¶

Bases: object

Lazy handle that resolves just the prompt/instruction.

Parameters:

runtime_prompt_id (str)
cache (RuntimeConfigCache | None)

property config: AgentRuntimeConfig | None¶

property prompt: str | None¶

refresh_callback(*args, **kwargs)[source]¶

Parameters:

args (Any)
kwargs (Any)

Return type:

None

class taimoe.platform.adapters.adk.handle.TaimoeHandle[source]¶

Bases: object

Global facade used by application code.

Most callers should use bind_to() to wire this handle to a taimoe.platform.registry.TaimoeRegistry in one call. bind_cache() / bind_agent_syncer() are kept as lower-level knobs for tests and multi-tenant scenarios that need finer control.

bind_to(registry)[source]¶

Wire this global handle to a TaimoeRegistry.

Equivalent to calling bind_cache(registry.cache) and bind_agent_syncer(registry.agent_syncer) — kept as a single ergonomic call so application code doesn’t need to know about the wiring contract.

Parameters:: registry (Any)
Return type:: None

unbind()[source]¶

Detach the cache and syncer references.

Mainly useful in tests that construct multiple registries within one process; production code typically calls bind_to once at startup and never unbinds.

Return type:: None

bind_cache(cache)[source]¶

Parameters:: cache (RuntimeConfigCache)
Return type:: None

bind_agent_syncer(syncer)[source]¶

Wire in the runtime-less syncer so that every agent(name) call automatically subscribes that name to the background poller.

Parameters:: syncer (Any)
Return type:: None

agent(runtime_agent_id)[source]¶

Parameters:: runtime_agent_id (str)
Return type:: TaimoeAgentHandle

prompt(runtime_prompt_id)[source]¶

Parameters:: runtime_prompt_id (str)
Return type:: TaimoePromptHandle

init(*, runtime_id, base_url=None, platform_url=None, api_key=None, enabled=True, raise_on_export_error=False, timeout=10.0)[source]¶

Configure observability export for this process.

Parameters:

runtime_id (str)
base_url (str | None)
platform_url (str | None)
api_key (str | None)
enabled (bool)
raise_on_export_error (bool)
timeout (float)

Return type:

None

trace(name, **kwargs)[source]¶

Start an observability trace.

Parameters:

name (str)
kwargs (Any)

Return type:

Any

Observability¶

Agent runtime observability helpers.

class taimoe.platform.observability.ObservabilityConfig(runtime_id, base_url, api_key=None, enabled=True, raise_on_export_error=False, timeout=10.0)[source]

Bases: object

Runtime observability configuration.

Parameters:

runtime_id (str)
base_url (str)
api_key (str | None)
enabled (bool)
raise_on_export_error (bool)
timeout (float)

runtime_id: str

base_url: str

api_key: str | None = None

enabled: bool = True

raise_on_export_error: bool = False

timeout: float = 10.0

class taimoe.platform.observability.ObservabilityState[source]

Bases: object

Holds the active exporter client for process-wide instrumentation.

config: ObservabilityConfig | None

client: TaimoeClient | None

configure(*, runtime_id, base_url, api_key=None, enabled=True, raise_on_export_error=False, timeout=10.0)[source]

Parameters:

runtime_id (str)
base_url (str)
api_key (str | None)
enabled (bool)
raise_on_export_error (bool)
timeout (float)

Return type:

None

require_config()[source]

Return type:: ObservabilityConfig

export(batch)[source]

Parameters:: batch (SpanBatch)
Return type:: None

class taimoe.platform.observability.TaimoeSpan(*, trace, span_type, name, parent_span_id=None, input=None, attributes=None, agent_id=None)[source]

Bases: object

Context manager for one agent, LLM, tool, app, or policy operation.

Parameters:

trace (TaimoeTrace)
span_type (SpanType)
name (str)
parent_span_id (str | None)
input (Any)
attributes (dict[str, Any] | None)
agent_id (str | None)

output: Any

prompt_tokens: int | None

completion_tokens: int | None

cost_usd: float | None

error: str | None

set_input(value)[source]

Parameters:: value (Any)
Return type:: None

set_output(value)[source]

Parameters:: value (Any)
Return type:: None

set_attribute(key, value)[source]

Parameters:

key (str)
value (Any)

Return type:

None

set_usage(*, prompt_tokens=None, completion_tokens=None, cost_usd=None)[source]

Parameters:

prompt_tokens (int | None)
completion_tokens (int | None)
cost_usd (float | None)

Return type:

None

set_error(error)[source]

Parameters:: error (BaseException | str)
Return type:: None

finish()[source]

Return type:: SpanEvent

class taimoe.platform.observability.TaimoeTrace(name, *, trace_id=None, session_id=None, agent_id=None, team_id=None, attributes=None, state=<taimoe.platform.observability.trace.ObservabilityState object>)[source]

Bases: object

Context manager representing one agent run or conversation trace.

Parameters:

name (str)
trace_id (str | None)
session_id (str | None)
agent_id (str | None)
team_id (str | None)
attributes (dict[str, Any] | None)
state (ObservabilityState)

export_error: Exception | None

property spans: tuple[SpanEvent, ...]

span(span_type, *, name, input=None, attributes=None, agent_id=None)[source]

Parameters:

span_type (Literal['agent', 'llm', 'tool', 'policy', 'workflow', 'app', 'custom'])
name (str)
input (Any)
attributes (dict[str, Any] | None)
agent_id (str | None)

Return type:

TaimoeSpan

record_span(span)[source]

Parameters:: span (SpanEvent)
Return type:: None

flush()[source]

Return type:: None

taimoe.platform.observability.current_span()[source]

Return the active span for the current context.

Return type:: TaimoeSpan | None

taimoe.platform.observability.current_trace()[source]

Return the active trace for the current context.

Return type:: TaimoeTrace | None

taimoe.platform.observability.init(*, runtime_id, base_url=None, platform_url=None, api_key=None, enabled=True, raise_on_export_error=False, timeout=10.0)[source]

Configure process-wide observability export to the Taimoe Platform.

Parameters:

runtime_id (str)
base_url (str | None)
platform_url (str | None)
api_key (str | None)
enabled (bool)
raise_on_export_error (bool)
timeout (float)

Return type:

None

taimoe.platform.observability.trace(name, *, trace_id=None, session_id=None, agent_id=None, team_id=None, attributes=None)[source]

Start a trace for one agent run or conversation.

Parameters:

name (str)
trace_id (str | None)
session_id (str | None)
agent_id (str | None)
team_id (str | None)
attributes (dict[str, Any] | None)

Return type:

TaimoeTrace

Trace/span instrumentation for agent runtimes.

Trace and span IDs follow W3C Trace Context format (32-hex / 16-hex, no prefix) so they round-trip through any OTel-aware collector. When opentelemetry is installed, every Taimoe span is mirrored as an OTel span via _otel so downstream APMs pick the trace up automatically.

class taimoe.platform.observability.trace.ObservabilityConfig(runtime_id, base_url, api_key=None, enabled=True, raise_on_export_error=False, timeout=10.0)[source]¶

Bases: object

Runtime observability configuration.

Parameters:

runtime_id (str)
base_url (str)
api_key (str | None)
enabled (bool)
raise_on_export_error (bool)
timeout (float)

runtime_id: str¶

base_url: str¶

api_key: str | None = None¶

enabled: bool = True¶

raise_on_export_error: bool = False¶

timeout: float = 10.0¶

class taimoe.platform.observability.trace.ObservabilityState[source]¶

Bases: object

Holds the active exporter client for process-wide instrumentation.

config: ObservabilityConfig | None¶

client: TaimoeClient | None¶

configure(*, runtime_id, base_url, api_key=None, enabled=True, raise_on_export_error=False, timeout=10.0)[source]¶

Parameters:

runtime_id (str)
base_url (str)
api_key (str | None)
enabled (bool)
raise_on_export_error (bool)
timeout (float)

Return type:

None

require_config()[source]¶

Return type:: ObservabilityConfig

export(batch)[source]¶

Parameters:: batch (SpanBatch)
Return type:: None

class taimoe.platform.observability.trace.TaimoeTrace(name, *, trace_id=None, session_id=None, agent_id=None, team_id=None, attributes=None, state=<taimoe.platform.observability.trace.ObservabilityState object>)[source]¶

Bases: object

Context manager representing one agent run or conversation trace.

Parameters:

name (str)
trace_id (str | None)
session_id (str | None)
agent_id (str | None)
team_id (str | None)
attributes (dict[str, Any] | None)
state (ObservabilityState)

export_error: Exception | None¶

property spans: tuple[SpanEvent, ...]¶

span(span_type, *, name, input=None, attributes=None, agent_id=None)[source]¶

Parameters:

span_type (Literal['agent', 'llm', 'tool', 'policy', 'workflow', 'app', 'custom'])
name (str)
input (Any)
attributes (dict[str, Any] | None)
agent_id (str | None)

Return type:

TaimoeSpan

record_span(span)[source]¶

Parameters:: span (SpanEvent)
Return type:: None

flush()[source]¶

Return type:: None

class taimoe.platform.observability.trace.TaimoeSpan(*, trace, span_type, name, parent_span_id=None, input=None, attributes=None, agent_id=None)[source]¶

Bases: object

Context manager for one agent, LLM, tool, app, or policy operation.

Parameters:

trace (TaimoeTrace)
span_type (SpanType)
name (str)
parent_span_id (str | None)
input (Any)
attributes (dict[str, Any] | None)
agent_id (str | None)

output: Any¶

prompt_tokens: int | None¶

completion_tokens: int | None¶

cost_usd: float | None¶

error: str | None¶

set_input(value)[source]¶

Parameters:: value (Any)
Return type:: None

set_output(value)[source]¶

Parameters:: value (Any)
Return type:: None

set_attribute(key, value)[source]¶

Parameters:

key (str)
value (Any)

Return type:

None

set_usage(*, prompt_tokens=None, completion_tokens=None, cost_usd=None)[source]¶

Parameters:

prompt_tokens (int | None)
completion_tokens (int | None)
cost_usd (float | None)

Return type:

None

set_error(error)[source]¶

Parameters:: error (BaseException | str)
Return type:: None

finish()[source]¶

Return type:: SpanEvent

taimoe.platform.observability.trace.init(*, runtime_id, base_url=None, platform_url=None, api_key=None, enabled=True, raise_on_export_error=False, timeout=10.0)[source]¶

Configure process-wide observability export to the Taimoe Platform.

Parameters:

runtime_id (str)
base_url (str | None)
platform_url (str | None)
api_key (str | None)
enabled (bool)
raise_on_export_error (bool)
timeout (float)

Return type:

None

taimoe.platform.observability.trace.trace(name, *, trace_id=None, session_id=None, agent_id=None, team_id=None, attributes=None)[source]¶

Start a trace for one agent run or conversation.

Parameters:

name (str)
trace_id (str | None)
session_id (str | None)
agent_id (str | None)
team_id (str | None)
attributes (dict[str, Any] | None)

Return type:

TaimoeTrace

taimoe.platform.observability.trace.current_trace()[source]¶

Return the active trace for the current context.

Return type:: TaimoeTrace | None

taimoe.platform.observability.trace.current_span()[source]¶

Return the active span for the current context.

Return type:: TaimoeSpan | None

Instrumentation¶

Auto-instrumentation modules.

taimoe.platform.instrumentation.patch_adk()¶

Apply every supported monkey-patch. Returns True if any succeeded.

Return type:: bool

taimoe.platform.instrumentation.unpatch_adk()¶

Reverse whatever patch() managed to apply.

Return type:: None

taimoe.platform.instrumentation.patch_all()[source]¶

Apply every supported auto-instrumentation.

Returns a {library: succeeded} map so callers can log or surface which integrations took effect. True means at least one of that library’s monkey-patches landed; False means the library wasn’t importable or every patch refused.

Return type:: dict[str, bool]

taimoe.platform.instrumentation.unpatch_all()[source]¶

Reverse whatever patch_all() managed to apply.

Return type:: None

taimoe.platform.instrumentation.is_adk_available()[source]¶

Return True if google.adk has already been imported by the host.

Used by init() so L4 / L5 callers don’t pay the ADK patch cost when they’re not using ADK. We check sys.modules rather than import google.adk because importing would itself pull ADK into the host’s import graph.

Return type:: bool

Auto-instrumentation for Google ADK.

The original implementation crammed three independent monkey-patches into a single patch() body, which made partial failures hard to recover from (a successful first patch left _patched=True even when later patches blew up, and unpatch() would then try to undo work that never happened). This module now exposes three small patch helpers and patch() simply orchestrates them with isolated state.

taimoe.platform.instrumentation.adk.patch()[source]¶

Apply every supported monkey-patch. Returns True if any succeeded.

Return type:: bool

taimoe.platform.instrumentation.adk.unpatch()[source]¶

Reverse whatever patch() managed to apply.

Return type:: None

Resources¶

Agents resource for Taimoe Platform SDK.

class taimoe.platform.resources.agents.AgentsResource(client)[source]¶

Bases: object

Resource for interacting with /api/v1/agents endpoints.

Parameters:: client (TaimoeClient)

get_runtime_config_by_name(name)[source]¶

Fetch a single agent’s runtime config by runtime_agent_id (or name). Runtime-less: no Runtime binding required.

Parameters:: name (str)
Return type:: AgentRuntimeConfig

async get_runtime_config_by_name_async(name)[source]¶

Parameters:: name (str)
Return type:: AgentRuntimeConfig

create(payload, *, knowledge_bases=None, org=None, team='default')[source]¶

Create a new agent and optionally bind knowledge bases to it.

Backend POST /organizations/{org}/teams/{team}/agents does not accept knowledge bases in the create body, so when knowledge_bases is provided the SDK issues one bind call per slug after creation.

Parameters:

payload (AgentCreatePayload) – Typed agent create payload.
knowledge_bases (list[str] | None) – Optional list of ‘team/kb’ slugs to bind.
org (str | None) – Organization slug. Falls back to the client’s org.
team (str) – Team slug.

Return type:

Agent

async create_async(payload, *, knowledge_bases=None, org=None, team='default')[source]¶

Create a new agent asynchronously.

Parameters:

payload (AgentCreatePayload)
knowledge_bases (list[str] | None)
org (str | None)
team (str)

Return type:

Agent

bind_knowledge_base(agent_id, kb_slug)[source]¶

Bind a knowledge base to an agent using its ‘team/kb’ slug.

The backend resolves the slug via kb_resolver, so no org/team scoping is needed in the URL.

Parameters:

agent_id (str)
kb_slug (str)

Return type:

AgentBinding

async bind_knowledge_base_async(agent_id, kb_slug)[source]¶

Bind a knowledge base to an agent asynchronously.

Parameters:

agent_id (str)
kb_slug (str)

Return type:

AgentBinding

unbind_knowledge_base(agent_id, kb_id)[source]¶

Unbind a knowledge base from an agent by its kb UUID.

Backend responds with 204 No Content.

Parameters:

agent_id (str)
kb_id (str | UUID)

Return type:

None

async unbind_knowledge_base_async(agent_id, kb_id)[source]¶

Unbind a knowledge base from an agent asynchronously.

Parameters:

agent_id (str)
kb_id (str | UUID)

Return type:

None

Knowledge bases resource for Taimoe Platform SDK.

class taimoe.platform.resources.knowledge_bases.KnowledgeBasesResource(client)[source]¶

Bases: object

Resource for interacting with knowledge base endpoints.

Parameters:: client (TaimoeClient)

list(team, org=None)[source]¶

List knowledge bases for a team. Backend returns a bare array.

Parameters:

team (str)
org (str | None)

Return type:

list[KnowledgeBase]

async list_async(team, org=None)[source]¶

List knowledge bases for a team asynchronously.

Parameters:

team (str)
org (str | None)

Return type:

list[KnowledgeBase]

get(slug, org=None)[source]¶

Get a specific knowledge base by slug.

Backend only exposes GET-by-UUID, so we list the team and filter.

Parameters:

slug (str)
org (str | None)

Return type:

KnowledgeBase

async get_async(slug, org=None)[source]¶

Get a specific knowledge base asynchronously by slug.

Parameters:

slug (str)
org (str | None)

Return type:

KnowledgeBase

test_query(slug, query, org=None)[source]¶

Test a query against a knowledge base using its slug.

Parameters:

slug (str)
query (str)
org (str | None)

Return type:

QueryResult

async test_query_async(slug, query, org=None)[source]¶

Test a query against a knowledge base asynchronously using its slug.

Parameters:

slug (str)
query (str)
org (str | None)

Return type:

QueryResult

Observability resource for uploading runtime spans.

class taimoe.platform.resources.observability.ObservabilityResource(client)[source]¶

Bases: object

Resource for interacting with /api/v1/observability endpoints.

Parameters:: client (TaimoeClient)

submit_spans(batch)[source]¶

Synchronously upload a batch of runtime spans.

Parameters:: batch (SpanBatch)
Return type:: None

async submit_spans_async(batch)[source]¶

Asynchronously upload a batch of runtime spans.

Parameters:: batch (SpanBatch)
Return type:: None

submit_span(span)[source]¶

Upload one span as a single-span batch.

Parameters:: span (SpanEvent)
Return type:: None

async submit_span_async(span)[source]¶

Asynchronously upload one span as a single-span batch.

Parameters:: span (SpanEvent)
Return type:: None

Runtimes resource for Taimoe Platform SDK.

class taimoe.platform.resources.runtimes.RuntimesResource(client)[source]¶

Bases: object

Resource for interacting with /api/v1/runtimes endpoints.

Parameters:: client (TaimoeClient)

sync_agents(runtime_id, *, since=None)[source]¶

Synchronously pull agent configurations for this runtime.

Parameters:

runtime_id (str)
since (datetime | None)

Return type:

RuntimeSyncResponse

async sync_agents_async(runtime_id, *, since=None)[source]¶

Asynchronously pull agent configurations for this runtime.

Parameters:

runtime_id (str)
since (datetime | None)

Return type:

RuntimeSyncResponse

Types¶

Public Pydantic models used by the Taimoe Platform SDK.

class taimoe.platform.types.Agent(*, id, name, display_name, model_alias, runtime_type, is_active, description=None, instruction=None, runtime_agent_id=None, tools=(), generation_config=<factory>, tpm_limit=0, rpm_limit=0, create_time, update_time=None)[source]¶

Bases: BaseModel

Agent representation as returned by the backend.

Parameters:

id (str)
name (str)
display_name (str)
model_alias (str)
runtime_type (str)
is_active (bool)
description (str | None)
instruction (str | None)
runtime_agent_id (str | None)
tools (tuple[str, ...])
generation_config (dict[str, Any])
tpm_limit (int)
rpm_limit (int)
create_time (datetime)
update_time (datetime | None)

model_config = {'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

id: str¶

name: str¶

display_name: str¶

model_alias: str¶

runtime_type: str¶

is_active: bool¶

description: str | None¶

instruction: str | None¶

runtime_agent_id: str | None¶

tools: tuple[str, ...]¶

generation_config: dict[str, Any]¶

tpm_limit: int¶

rpm_limit: int¶

create_time: datetime¶

update_time: datetime | None¶

class taimoe.platform.types.AgentBinding(*, kb_id, agent_id, slug, name, source_type, display_name=None, grounding_source=None, is_active, bound_at)[source]¶

Bases: BaseModel

Result of binding a knowledge base to an agent.

Parameters:

kb_id (str)
agent_id (str)
slug (str)
name (str)
source_type (str)
display_name (str | None)
grounding_source (str | None)
is_active (bool)
bound_at (datetime)

model_config = {'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

kb_id: str¶

agent_id: str¶

slug: str¶

name: str¶

source_type: str¶

display_name: str | None¶

grounding_source: str | None¶

is_active: bool¶

bound_at: datetime¶

class taimoe.platform.types.AgentCreatePayload(*, name, model_alias, description=None, runtime_type='google_agent_engine', is_active=True, instruction=None, runtime_agent_id=None, runtime_id=None, tools=None, generation_config=None, tpm_limit=0, rpm_limit=0)[source]¶

Bases: BaseModel

Typed payload for creating an agent.

Mirrors the backend AgentCreate Pydantic schema; SDK callers get IDE completion and typo protection instead of **kwargs: Any.

Serialization contract — the resources layer dumps this with model_dump(exclude_none=True) so optional fields left as None aren’t sent on the wire; backend defaults take over. If you set a field to its falsy value explicitly (e.g. tools=()), it will be sent, since the value isn’t None.

Parameters:

name (str)
model_alias (str)
description (str | None)
runtime_type (str)
is_active (bool)
instruction (str | None)
runtime_agent_id (str | None)
runtime_id (str | None)
tools (tuple[str, ...] | None)
generation_config (dict[str, Any] | None)
tpm_limit (int)
rpm_limit (int)

model_config = {'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str¶

model_alias: str¶

description: str | None¶

runtime_type: str¶

is_active: bool¶

instruction: str | None¶

runtime_agent_id: str | None¶

runtime_id: str | None¶

tools: tuple[str, ...] | None¶

generation_config: dict[str, Any] | None¶

tpm_limit: int¶

rpm_limit: int¶

class taimoe.platform.types.AgentManifest(*, id, name=None, kind=None, entrypoint=True, sub_agents=(), declared_tools=(), metadata=<factory>)[source]¶

Bases: BaseModel

Code-level agent shape discovered from the runtime.

This intentionally avoids platform-managed prompt or model config. Platform config is synchronized separately through the sync API.

Parameters:

id (str)
name (str | None)
kind (str | None)
entrypoint (bool)
sub_agents (tuple[str, ...])
declared_tools (tuple[str, ...])
metadata (dict[str, Any])

model_config = {'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

id: str¶

name: str | None¶

kind: str | None¶

entrypoint: bool¶

sub_agents: tuple[str, ...]¶

declared_tools: tuple[str, ...]¶

metadata: dict[str, Any]¶

class taimoe.platform.types.AgentRuntimeConfig(*, runtime_agent_id, version, instruction=None, model_alias=None, generation_config=<factory>, enable_google_search=False, knowledge_base_ids=(), tools=())[source]¶

Bases: BaseModel

Resolved runtime config for a single agent.

The platform owns composition. The SDK caches and applies this shape.

Parameters:

runtime_agent_id (str)
version (int)
instruction (str | None)
model_alias (str | None)
generation_config (dict[str, Any])
enable_google_search (bool)
knowledge_base_ids (tuple[str, ...])
tools (tuple[str, ...])

model_config = {'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

runtime_agent_id: str¶

version: int¶

instruction: str | None¶

model_alias: str | None¶

generation_config: dict[str, Any]¶

enable_google_search: bool¶

knowledge_base_ids: tuple[str, ...]¶

tools: tuple[str, ...]¶

class taimoe.platform.types.AgentsManifest(*, agents)[source]¶

Bases: BaseModel

List response for the runtime agents discovery endpoint.

Parameters:: agents (tuple[AgentManifest, ...])

model_config = {'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

agents: tuple[AgentManifest, ...]¶

class taimoe.platform.types.HealthStatus(*, status='healthy', version, uptime_seconds)[source]¶

Bases: BaseModel

Runtime health response.

Parameters:

status (Literal['healthy', 'degraded', 'unhealthy'])
version (str)
uptime_seconds (float)

model_config = {'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

status: Literal['healthy', 'degraded', 'unhealthy']¶

version: str¶

uptime_seconds: float¶

class taimoe.platform.types.KnowledgeBase(*, id, slug, name, description=None)[source]¶

Bases: BaseModel

Knowledge base representation.

Parameters:

id (str)
slug (str)
name (str)
description (str | None)

model_config = {'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

id: str¶

slug: str¶

name: str¶

description: str | None¶

class taimoe.platform.types.QueryResult(*, query, answer=None, chunks=[])[source]¶

Bases: BaseModel

Result from testing a query against a KB.

Parameters:

query (str)
answer (str | None)
chunks (list[dict[str, Any]])

model_config = {'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

query: str¶

answer: str | None¶

chunks: list[dict[str, Any]]¶

class taimoe.platform.types.RuntimeManifest(*, taimoe_protocol_version='1.0', runtime_name, runtime_type='api', sdk_version, framework=None, framework_version=None, started_at=<factory>, endpoints=<factory>)[source]¶

Bases: BaseModel

Metadata used by Taimoe Platform to identify a runtime service.

Parameters:

taimoe_protocol_version (str)
runtime_name (str)
runtime_type (Literal['api'])
sdk_version (str)
framework (str | None)
framework_version (str | None)
started_at (datetime)
endpoints (RuntimeEndpoints)

model_config = {'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

taimoe_protocol_version: str¶

runtime_name: str¶

runtime_type: Literal['api']¶

sdk_version: str¶

framework: str | None¶

framework_version: str | None¶

started_at: datetime¶

endpoints: RuntimeEndpoints¶

class taimoe.platform.types.RuntimeSyncResponse(*, synced_at, agents=())[source]¶

Bases: BaseModel

Response returned by the platform sync endpoint.

Parameters:

synced_at (datetime)
agents (tuple[AgentRuntimeConfig, ...])

model_config = {'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

synced_at: datetime¶

agents: tuple[AgentRuntimeConfig, ...]¶

class taimoe.platform.types.SpanBatch(*, runtime_id, trace_id, session_id=None, spans=())[source]¶

Bases: BaseModel

Batch upload payload for runtime observability spans.

Parameters:

runtime_id (str)
trace_id (str)
session_id (str | None)
spans (tuple[SpanEvent, ...])

model_config = {'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

runtime_id: str¶

trace_id: str¶

session_id: str | None¶

spans: tuple[SpanEvent, ...]¶

class taimoe.platform.types.SpanEvent(*, trace_id, span_id, parent_span_id=None, session_id=None, runtime_id, agent_id=None, team_id=None, span_type='custom', name, status='ok', start_time, end_time, latency_ms, input=None, output=None, error=None, prompt_tokens=None, completion_tokens=None, cost_usd=None, attributes=<factory>)[source]¶

Bases: BaseModel

A single trace span emitted by a managed runtime.

Parameters:

trace_id (str)
span_id (str)
parent_span_id (str | None)
session_id (str | None)
runtime_id (str)
agent_id (str | None)
team_id (str | None)
span_type (Literal['agent', 'llm', 'tool', 'policy', 'workflow', 'app', 'custom'])
name (str)
status (Literal['ok', 'error'])
start_time (datetime)
end_time (datetime)
latency_ms (float)
input (Any)
output (Any)
error (str | None)
prompt_tokens (int | None)
completion_tokens (int | None)
cost_usd (float | None)
attributes (dict[str, Any])

model_config = {'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

trace_id: str¶

span_id: str¶

parent_span_id: str | None¶

session_id: str | None¶

runtime_id: str¶

agent_id: str | None¶

team_id: str | None¶

span_type: SpanType¶

name: str¶

status: SpanStatus¶

start_time: datetime¶

end_time: datetime¶

latency_ms: float¶

input: Any¶

output: Any¶

error: str | None¶

prompt_tokens: int | None¶

completion_tokens: int | None¶

cost_usd: float | None¶

attributes: dict[str, Any]¶

Knowledge base data types.

class taimoe.platform.types.knowledge_base.KnowledgeBase(*, id, slug, name, description=None)[source]¶

Bases: BaseModel

Knowledge base representation.

Parameters:

id (str)
slug (str)
name (str)
description (str | None)

model_config = {'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

id: str¶

slug: str¶

name: str¶

description: str | None¶

class taimoe.platform.types.knowledge_base.QueryResult(*, query, answer=None, chunks=[])[source]¶

Bases: BaseModel

Result from testing a query against a KB.

Parameters:

query (str)
answer (str | None)
chunks (list[dict[str, Any]])

model_config = {'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

query: str¶

answer: str | None¶

chunks: list[dict[str, Any]]¶

Runtime discovery manifest models.

class taimoe.platform.types.manifest.RuntimeEndpoints(*, health='/health', agents='/agents', well_known='/.well-known/taimoe-runtime.json', invoke=None)[source]¶

Bases: BaseModel

Endpoint paths exposed by a managed runtime service.

Parameters:

health (str)
agents (str)
well_known (str)
invoke (str | None)

model_config = {'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

health: str¶

agents: str¶

well_known: str¶

invoke: str | None¶

class taimoe.platform.types.manifest.RuntimeManifest(*, taimoe_protocol_version='1.0', runtime_name, runtime_type='api', sdk_version, framework=None, framework_version=None, started_at=<factory>, endpoints=<factory>)[source]¶

Bases: BaseModel

Metadata used by Taimoe Platform to identify a runtime service.

Parameters:

taimoe_protocol_version (str)
runtime_name (str)
runtime_type (Literal['api'])
sdk_version (str)
framework (str | None)
framework_version (str | None)
started_at (datetime)
endpoints (RuntimeEndpoints)

model_config = {'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

taimoe_protocol_version: str¶

runtime_name: str¶

runtime_type: Literal['api']¶

sdk_version: str¶

framework: str | None¶

framework_version: str | None¶

started_at: datetime¶

endpoints: RuntimeEndpoints¶

class taimoe.platform.types.manifest.AgentManifest(*, id, name=None, kind=None, entrypoint=True, sub_agents=(), declared_tools=(), metadata=<factory>)[source]¶

Bases: BaseModel

Code-level agent shape discovered from the runtime.

This intentionally avoids platform-managed prompt or model config. Platform config is synchronized separately through the sync API.

Parameters:

id (str)
name (str | None)
kind (str | None)
entrypoint (bool)
sub_agents (tuple[str, ...])
declared_tools (tuple[str, ...])
metadata (dict[str, Any])

model_config = {'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

id: str¶

name: str | None¶

kind: str | None¶

entrypoint: bool¶

sub_agents: tuple[str, ...]¶

declared_tools: tuple[str, ...]¶

metadata: dict[str, Any]¶

class taimoe.platform.types.manifest.AgentsManifest(*, agents)[source]¶

Bases: BaseModel

List response for the runtime agents discovery endpoint.

Parameters:: agents (tuple[AgentManifest, ...])

model_config = {'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

agents: tuple[AgentManifest, ...]¶

class taimoe.platform.types.manifest.HealthStatus(*, status='healthy', version, uptime_seconds)[source]¶

Bases: BaseModel

Runtime health response.

Parameters:

status (Literal['healthy', 'degraded', 'unhealthy'])
version (str)
uptime_seconds (float)

model_config = {'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

status: Literal['healthy', 'degraded', 'unhealthy']¶

version: str¶

uptime_seconds: float¶

Observability payload models sent from runtime SDKs to the platform.

Trace and span IDs are W3C Trace Context format (lowercase hex). Pydantic validators reject anything else at the boundary, so server-side audit storage can rely on the format invariant without re-checking.

class taimoe.platform.types.observability.SpanEvent(*, trace_id, span_id, parent_span_id=None, session_id=None, runtime_id, agent_id=None, team_id=None, span_type='custom', name, status='ok', start_time, end_time, latency_ms, input=None, output=None, error=None, prompt_tokens=None, completion_tokens=None, cost_usd=None, attributes=<factory>)[source]¶

Bases: BaseModel

A single trace span emitted by a managed runtime.

Parameters:

trace_id (str)
span_id (str)
parent_span_id (str | None)
session_id (str | None)
runtime_id (str)
agent_id (str | None)
team_id (str | None)
span_type (Literal['agent', 'llm', 'tool', 'policy', 'workflow', 'app', 'custom'])
name (str)
status (Literal['ok', 'error'])
start_time (datetime)
end_time (datetime)
latency_ms (float)
input (Any)
output (Any)
error (str | None)
prompt_tokens (int | None)
completion_tokens (int | None)
cost_usd (float | None)
attributes (dict[str, Any])

model_config = {'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

trace_id: str¶

span_id: str¶

parent_span_id: str | None¶

session_id: str | None¶

runtime_id: str¶

agent_id: str | None¶

team_id: str | None¶

span_type: SpanType¶

name: str¶

status: SpanStatus¶

start_time: datetime¶

end_time: datetime¶

latency_ms: float¶

input: Any¶

output: Any¶

error: str | None¶

prompt_tokens: int | None¶

completion_tokens: int | None¶

cost_usd: float | None¶

attributes: dict[str, Any]¶

class taimoe.platform.types.observability.SpanBatch(*, runtime_id, trace_id, session_id=None, spans=())[source]¶

Bases: BaseModel

Batch upload payload for runtime observability spans.

Parameters:

runtime_id (str)
trace_id (str)
session_id (str | None)
spans (tuple[SpanEvent, ...])

model_config = {'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

runtime_id: str¶

trace_id: str¶

session_id: str | None¶

spans: tuple[SpanEvent, ...]¶

Platform-to-runtime sync models.

class taimoe.platform.types.sync.AgentRuntimeConfig(*, runtime_agent_id, version, instruction=None, model_alias=None, generation_config=<factory>, enable_google_search=False, knowledge_base_ids=(), tools=())[source]¶

Bases: BaseModel

Resolved runtime config for a single agent.

The platform owns composition. The SDK caches and applies this shape.

Parameters:

runtime_agent_id (str)
version (int)
instruction (str | None)
model_alias (str | None)
generation_config (dict[str, Any])
enable_google_search (bool)
knowledge_base_ids (tuple[str, ...])
tools (tuple[str, ...])

model_config = {'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

runtime_agent_id: str¶

version: int¶

instruction: str | None¶

model_alias: str | None¶

generation_config: dict[str, Any]¶

enable_google_search: bool¶

knowledge_base_ids: tuple[str, ...]¶

tools: tuple[str, ...]¶

class taimoe.platform.types.sync.RuntimeSyncResponse(*, synced_at, agents=())[source]¶

Bases: BaseModel

Response returned by the platform sync endpoint.

Parameters:

synced_at (datetime)
agents (tuple[AgentRuntimeConfig, ...])

model_config = {'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

synced_at: datetime¶

agents: tuple[AgentRuntimeConfig, ...]¶

Utilities¶

Function-tracking decorators (sync and async aware).

track_action wraps an individual tool call inside the current trace; no trace is auto-created. track_agent wraps an agent entry point and will mint a trace if the caller doesn’t already have one open.

Both flavors detect async def callables via inspect and produce a correctly-shaped wrapper, so callers don’t need separate track_action_async or track_agent_async symbols.

taimoe.platform.decorators.track_action(name=None)[source]¶

Wrap a function as a tool span on the active trace.

Parameters:: name (str | None)
Return type:: Callable[[F], F]

taimoe.platform.decorators.track_agent(name=None)[source]¶

Wrap an agent entry point as an agent span, minting a trace if needed.

Works for sync and async def callables.

Parameters:: name (str | None)
Return type:: Callable[[F], F]

Canonical errors for the Taimoe Platform SDK.

All errors derive from TaimoeError. Errors raised in response to an API call derive from TaimoeAPIError; errors caused by SDK-side misconfiguration derive from ConfigurationError.

Retry policies should branch on the _Retryable marker rather than hard-coded status codes.

exception taimoe.platform.errors.TaimoeError[source]¶

Bases: Exception

Base for every error raised by the Taimoe Platform SDK.

exception taimoe.platform.errors.TaimoeAPIError(message, *, status_code=0, code=None, request_id=None, details=None, response=None)[source]¶

Bases: TaimoeError

Raised when an API request to the Taimoe Platform fails.

Parameters:

message (str)
status_code (int)
code (str | None)
request_id (str | None)
details (dict[str, Any] | None)
response (httpx.Response | None)

Return type:

None

status_code¶: HTTP status from the response (or 0 for transport-level).

code¶: Canonical logical code (e.g. "RESOURCE_EXHAUSTED") if the server returned a structured detail, else None.

message¶: Human-readable message extracted from the response.

request_id¶: X-Request-ID echoed by the server, for log correlation.

details¶: Any additional fields from the structured detail body.

response¶: The raw httpx Response, kept for advanced debugging.

default_code: ClassVar[str | None] = None¶: Default canonical code for the subclass. Overridden by concrete classes.

classmethod from_response(response)[source]¶

Build the most specific subclass for an HTTP error response.

Resolution order:

structured detail.code → _CODE_REGISTRY
HTTP status → _STATUS_REGISTRY
fall back to TaimoeAPIError

Parameters:: response (httpx.Response)
Return type:: TaimoeAPIError

exception taimoe.platform.errors.ConfigurationError[source]¶

Bases: TaimoeError

The SDK was constructed or used with invalid configuration.

exception taimoe.platform.errors.InvalidArgumentError(message, *, status_code=0, code=None, request_id=None, details=None, response=None)[source]¶

Bases: TaimoeAPIError

The request was malformed or contained invalid arguments.

Parameters:

message (str)
status_code (int)
code (str | None)
request_id (str | None)
details (dict[str, Any] | None)
response (httpx.Response | None)

Return type:

None

default_code: ClassVar[str | None] = 'INVALID_ARGUMENT'¶: Default canonical code for the subclass. Overridden by concrete classes.

exception taimoe.platform.errors.FailedPreconditionError(message, *, status_code=0, code=None, request_id=None, details=None, response=None)[source]¶

Bases: TaimoeAPIError

Operation rejected because the system state didn’t satisfy a precondition.

Parameters:

message (str)
status_code (int)
code (str | None)
request_id (str | None)
details (dict[str, Any] | None)
response (httpx.Response | None)

Return type:

None

default_code: ClassVar[str | None] = 'FAILED_PRECONDITION'¶: Default canonical code for the subclass. Overridden by concrete classes.

exception taimoe.platform.errors.UnauthenticatedError(message, *, status_code=0, code=None, request_id=None, details=None, response=None)[source]¶

Bases: TaimoeAPIError

The request lacked valid authentication credentials.

Parameters:

message (str)
status_code (int)
code (str | None)
request_id (str | None)
details (dict[str, Any] | None)
response (httpx.Response | None)

Return type:

None

default_code: ClassVar[str | None] = 'UNAUTHENTICATED'¶: Default canonical code for the subclass. Overridden by concrete classes.

exception taimoe.platform.errors.PermissionDeniedError(message, *, status_code=0, code=None, request_id=None, details=None, response=None)[source]¶

Bases: TaimoeAPIError

The caller was authenticated but not authorized for this resource.

Parameters:

message (str)
status_code (int)
code (str | None)
request_id (str | None)
details (dict[str, Any] | None)
response (httpx.Response | None)

Return type:

None

default_code: ClassVar[str | None] = 'PERMISSION_DENIED'¶: Default canonical code for the subclass. Overridden by concrete classes.

exception taimoe.platform.errors.NotFoundError(message, *, status_code=0, code=None, request_id=None, details=None, response=None)[source]¶

Bases: TaimoeAPIError

The requested resource does not exist.

Parameters:

message (str)
status_code (int)
code (str | None)
request_id (str | None)
details (dict[str, Any] | None)
response (httpx.Response | None)

Return type:

None

default_code: ClassVar[str | None] = 'NOT_FOUND'¶: Default canonical code for the subclass. Overridden by concrete classes.

exception taimoe.platform.errors.AlreadyExistsError(message, *, status_code=0, code=None, request_id=None, details=None, response=None)[source]¶

Bases: TaimoeAPIError

The resource the caller tried to create already exists.

Parameters:

message (str)
status_code (int)
code (str | None)
request_id (str | None)
details (dict[str, Any] | None)
response (httpx.Response | None)

Return type:

None

default_code: ClassVar[str | None] = 'ALREADY_EXISTS'¶: Default canonical code for the subclass. Overridden by concrete classes.

exception taimoe.platform.errors.RateLimitError(message, *, status_code=0, code=None, request_id=None, details=None, response=None)[source]¶

Bases: TaimoeAPIError, _Retryable

Quota or rate limit was exhausted for this caller.

Parameters:

message (str)
status_code (int)
code (str | None)
request_id (str | None)
details (dict[str, Any] | None)
response (httpx.Response | None)

Return type:

None

default_code: ClassVar[str | None] = 'RESOURCE_EXHAUSTED'¶: Default canonical code for the subclass. Overridden by concrete classes.

property retry_after_seconds: int | None¶: Seconds the server asked the caller to wait before retrying, if any.

exception taimoe.platform.errors.InternalError(message, *, status_code=0, code=None, request_id=None, details=None, response=None)[source]¶

Bases: TaimoeAPIError, _Retryable

The server hit an unexpected condition.

Parameters:

message (str)
status_code (int)
code (str | None)
request_id (str | None)
details (dict[str, Any] | None)
response (httpx.Response | None)

Return type:

None

default_code: ClassVar[str | None] = 'INTERNAL'¶: Default canonical code for the subclass. Overridden by concrete classes.

exception taimoe.platform.errors.ServiceUnavailableError(message, *, status_code=0, code=None, request_id=None, details=None, response=None)[source]¶

Bases: TaimoeAPIError, _Retryable

The service is temporarily unavailable. Safe to retry with backoff.

Parameters:

message (str)
status_code (int)
code (str | None)
request_id (str | None)
details (dict[str, Any] | None)
response (httpx.Response | None)

Return type:

None

default_code: ClassVar[str | None] = 'UNAVAILABLE'¶: Default canonical code for the subclass. Overridden by concrete classes.

exception taimoe.platform.errors.DeadlineExceededError(message, *, status_code=0, code=None, request_id=None, details=None, response=None)[source]¶

Bases: TaimoeAPIError, _Retryable

The operation didn’t complete before the deadline (server or client side).

Parameters:

message (str)
status_code (int)
code (str | None)
request_id (str | None)
details (dict[str, Any] | None)
response (httpx.Response | None)

Return type:

None

default_code: ClassVar[str | None] = 'DEADLINE_EXCEEDED'¶: Default canonical code for the subclass. Overridden by concrete classes.

L4 proxy-mode helpers.

The proxy package produces and consumes the wire-level headers that flow between SDK callers, the Taimoe Gateway, and downstream OTel-aware services.

Producers (clients / agents) use build_taimoe_headers(). Consumers (gateway / audit middleware / tests) use parse_taimoe_headers().

All trace/span IDs follow W3C Trace Context format so the same trace flows through OTel collectors without translation.

taimoe.platform.proxy.build_taimoe_headers(*, runtime_id, agent_id, trace_id=None, span_id=None, parent_span_id=None, parent_agent_id=None, step=None, session_id=None, user_id=None, sampled=True, include_traceparent=True)[source]¶

Compose Taimoe-native + W3C trace headers for an outbound HTTP call.

Parameters:

runtime_id (str) – ID of the runtime making the call. Required.
agent_id (str) – ID of the agent making the call. Required.
trace_id (str | None) – 32-hex trace ID. Generated if absent.
span_id (str | None) – 16-hex span ID for this call. Generated if absent.
parent_span_id (str | None) – 16-hex span ID of the caller, when known.
parent_agent_id (str | None) – Agent ID of the caller, when known.
step (str | None) – Lifecycle step label (e.g. "pre_call").
session_id (str | None) – End-user session correlation ID.
user_id (str | None) – End-user identity (already privacy-filtered upstream).
sampled (bool) – Whether to set the W3C sampled flag.
include_traceparent (bool) – When False, only emit X-Taimoe-* headers. Useful when the caller already manages W3C headers itself.

Returns:

A flat dict[str, str] suitable for httpx/requests.

Return type:

dict[str, str]

taimoe.platform.proxy.build_traceparent(trace_id, span_id, *, sampled=True)[source]¶

Build a W3C traceparent value from a (trace_id, span_id) pair.

Parameters:

trace_id (str) – 32 lowercase hex chars.
span_id (str) – 16 lowercase hex chars.
sampled (bool) – Whether the trace should be exported by downstream APMs.

Raises:

ValueError – if either ID is not in W3C format.

Return type:

str

taimoe.platform.proxy.build_tracestate(*, runtime_id=None)[source]¶

Build a minimal tracestate carrying our vendor segment.

For now we only stamp the runtime; we can add more vendor keys later without breaking parsers (downstream services preserve unknown vendors).

Parameters:: runtime_id (str | None)
Return type:: str

taimoe.platform.proxy.parse_taimoe_headers(headers)[source]¶

Reconstruct request context from inbound headers.

Prefers Taimoe-native X-Taimoe-* headers; if they’re missing but traceparent is present, the trace and span IDs are pulled from it. Returns a context with all-None fields when nothing matches — callers can then decide whether to mint fresh IDs.

Parameters:: headers (Mapping[str, str])
Return type:: TaimoeRequestContext

taimoe.platform.proxy.parse_traceparent(value)[source]¶

Parse a W3C traceparent value, returning None if malformed.

Format: version-traceid-spanid-flags (all hex). Per spec, unknown future versions MUST still parse the first three fields.

Parameters:: value (str)
Return type:: TraceParent | None

class taimoe.platform.proxy.TaimoeRequestContext(trace_id, span_id, parent_span_id, runtime_id, agent_id, parent_agent_id, step, session_id, user_id)[source]¶

Bases: object

Result of parsing a request’s Taimoe-native headers.

Parameters:

trace_id (str | None)
span_id (str | None)
parent_span_id (str | None)
runtime_id (str | None)
agent_id (str | None)
parent_agent_id (str | None)
step (str | None)
session_id (str | None)
user_id (str | None)

trace_id: str | None¶

span_id: str | None¶

parent_span_id: str | None¶

runtime_id: str | None¶

agent_id: str | None¶

parent_agent_id: str | None¶

step: str | None¶

session_id: str | None¶

user_id: str | None¶

class taimoe.platform.proxy.TraceParent(version, trace_id, span_id, sampled)[source]¶

Bases: object

Parsed W3C traceparent value.

Parameters:

version (str)
trace_id (str)
span_id (str)
sampled (bool)

version: str¶

trace_id: str¶

span_id: str¶

sampled: bool¶

Optional FastAPI routes for runtime discovery.

Exposes three endpoints a managed runtime advertises so the Taimoe Platform can probe it:

GET /.well-known/taimoe-runtime.json — RFC 8615 well-known descriptor. Carries the runtime manifest and optionally answers a challenge so the Platform can verify the runtime is running an actual Taimoe SDK.
GET /agents — list of agents currently registered at this runtime.
GET /health — runtime health status (uptime, version, etc).

Per RFC 8615 the well-known endpoint must live at the URL root. To make that hard to get wrong we return two routers — one anchored at root, one that can be mounted at any prefix — and the helper wires both into a FastAPI app for you.

taimoe.platform.routes.compute_challenge_response(challenge)[source]¶

Hash challenge with the protocol version for the well-known reply.

Anti-spoofing only — see _well_known.CHALLENGE_PROTOCOL_VERSION for why this isn’t authentication and how to evolve it.

Parameters:: challenge (str)
Return type:: str

taimoe.platform.routes.create_well_known_router(registry)[source]¶

Router carrying only GET /.well-known/taimoe-runtime.json.

Must be mounted at the FastAPI app root with no prefix — RFC 8615 requires the well-known URI to be served from origin root.

Parameters:: registry (TaimoeRegistry)
Return type:: APIRouter

taimoe.platform.routes.create_discovery_router(registry)[source]¶

Router carrying /agents and /health.

Can be mounted at any prefix the host application prefers (e.g. prefix="/v1").

Parameters:: registry (TaimoeRegistry)
Return type:: APIRouter

taimoe.platform.routes.install_routes(registry, app)[source]¶

Wire both routers into a FastAPI app with the correct mounts.

The well-known router goes at the root; the discovery router is mounted bare (/agents / /health). Callers that need a different prefix for discovery can use create_discovery_router() directly.

Parameters:

registry (TaimoeRegistry)
app (FastAPI)

Return type:

None

taimoe.platform.routes.create_fastapi_router(registry)[source]¶

Backwards-compatible single-router entry point.

Returns one combined router that includes both the well-known and discovery routes. Convenient for simple deployments, but means the caller must mount it at the app root or the well-known path will break RFC 8615. Prefer install_routes() for new code.

Parameters:: registry (TaimoeRegistry)
Return type:: APIRouter