Remote client (Tier 4)

Drives a hosted Kendra Orchestrator engine over HTTP. Optional — nothing in Tiers 1-3 depends on it. Requires the [remote] extra.

pip install "korchestrator[remote]"

KorchestratorClient

KorchestratorClient(
    base_url: str,
    *,
    api_key: str | None = None,
    timeout: float = DEFAULT_TIMEOUT_SECONDS,
    max_retries: int = DEFAULT_MAX_RETRIES,
)

Thin client for a hosted Korchestrator engine (spec 04 §7, Tier 4, the [remote] extra).

Speaks the documented remote contract over HTTP: one Authorization: Bearer header for both a static API key and a Keycloak/KIAM JWT (spec 04 §7.2, ADR 0005), a 30s default per-request timeout (overridable per call), and up to 3 retries with full-jitter exponential backoff on 429/502/503/504 and connection failures — never on any other 4xx, since retrying a client error is a defect (spec 04 §7.5). Covers run lifecycle (run/run_swarm/ get_run/wait/run_and_wait/list_runs/get_run_summary), control and identity (resume/cancel/edit_resume/me/my_quota/my_runs/key management), discovery (tools/models/swarm_templates), and live event streaming (stream, a native async iterator) — every method except stream is a sync wrapper around an async core (spec 04 §7's own Tier-4 example calls client.run_and_wait(...) with no await). Nothing in Tiers 1-3 (Korch/Swarm/the kernel) depends on this client, and the base install never imports it.

Example

from korchestrator.remote import KorchestratorClient client = KorchestratorClient("https://engine.example.com", api_key="sk-example") client.close()

Build a client bound to base_url, authenticating with api_key when given.

Parameters:

Name	Type	Description	Default
base_url	str	The engine's base URL, e.g. "https://engine.example.com".	required
api_key	str | None	A static per-tenant key or a Keycloak/KIAM JWT — both ride the same Authorization: Bearer header (spec 04 §7.2). Omit for an unauthenticated engine (local development only).	None
timeout	float	The default per-request timeout in seconds; overridable per call.	DEFAULT_TIMEOUT_SECONDS
max_retries	int	How many times a retryable failure is retried before raising.	DEFAULT_MAX_RETRIES

__aenter__ async

__aenter__() -> Self

Support async with KorchestratorClient(...) as client:.

__aexit__ async

__aexit__(
    exc_type: type[BaseException] | None,
    exc: BaseException | None,
    traceback: TracebackType | None,
) -> None

Close the underlying connection pool on exit.

__repr__

__repr__() -> str

Show base_url only — never the credential (spec 04 §7.2: never logged/printed).

aclose async

aclose() -> None

Close the underlying connection pool (async).

cancel

cancel(
    run_id: str, *, timeout: float | None = None
) -> RemoteRunResult

Cancel a run (POST /v1/run/{id}/cancel).

Parameters:

Name	Type	Description	Default
run_id	str	The run to cancel.	required
timeout	float | None	Overrides the client's default timeout for this call only.	None

Returns:

Type	Description
RemoteRunResult	The run's :class:~korchestrator.models.remote.RemoteRunResult after cancelling.

Raises:

Type	Description
ApiError	The engine rejected the request or the run does not exist.
NetworkError	The connection failed after retries were exhausted.
TimeoutError	The request timed out after retries were exhausted.

Example

from korchestrator.remote import KorchestratorClient client = KorchestratorClient("https://engine.example.com", api_key="sk-example") client.cancel("r1") # doctest: +SKIP client.close()

close

close() -> None

Close the underlying connection pool (sync convenience wrapper around :meth:aclose).

create_key

create_key(
    *,
    scopes: Sequence[str] = (),
    timeout: float | None = None,
) -> ApiKey

Create a new API key (POST /v1/keys, the korchestrator:admin scope).

The returned secret is shown once — the engine never returns it again.

Parameters:

Name	Type	Description	Default
scopes	Sequence[str]	The scopes to grant the new key.	()
timeout	float | None	Overrides the client's default timeout for this call only.	None

Returns:

Type	Description
ApiKey	The newly created :class:~korchestrator.models.remote.ApiKey.

Raises:

Type	Description
ApiError	The engine rejected the request (e.g. insufficient scope).
NetworkError	The connection failed after retries were exhausted.
TimeoutError	The request timed out after retries were exhausted.

Example

from korchestrator.remote import KorchestratorClient client = KorchestratorClient("https://engine.example.com", api_key="sk-example") client.create_key(scopes=["korchestrator:read"]) # doctest: +SKIP client.close()

edit_resume

edit_resume(
    run_id: str,
    *,
    updates: Mapping[str, JSONValue] | None = None,
    trust_delta: float = 0.0,
    timeout: float | None = None,
) -> RemoteRunResult

Modify a governance_paused run's state and resume it (POST .../edit-resume).

Mirrors the local kernel's own edit_resume signal (services.Korch.edit_resume): updates are merged last-value into the run's context channel, and trust_delta is folded into trust_score.

Parameters:

Name	Type	Description	Default
run_id	str	The paused run to edit and resume.	required
updates	Mapping[str, JSONValue] | None	Context-channel values to merge (last-value) into the paused run's state.	None
trust_delta	float	Folded into trust_score by the engine's own clamp.	0.0
timeout	float | None	Overrides the client's default timeout for this call only.	None

Returns:

Type	Description
RemoteRunResult	The run's :class:~korchestrator.models.remote.RemoteRunResult after resuming.

Raises:

Type	Description
ApiError	The engine rejected the request or the run is not paused.
NetworkError	The connection failed after retries were exhausted.
TimeoutError	The request timed out after retries were exhausted.

Example

from korchestrator.remote import KorchestratorClient client = KorchestratorClient("https://engine.example.com", api_key="sk-example") client.edit_resume("r1", trust_delta=0.1) # doctest: +SKIP client.close()

get_run

get_run(
    run_id: str, *, timeout: float | None = None
) -> RemoteRunResult

Fetch a run's full live state (GET /v1/run/{id}).

Parameters:

Name	Type	Description	Default
run_id	str	The run to fetch.	required
timeout	float | None	Overrides the client's default timeout for this call only.	None

Returns:

Type	Description
RemoteRunResult	The run's current :class:~korchestrator.models.remote.RemoteRunResult.

Raises:

Type	Description
ApiError	The engine rejected the request or the run does not exist.
NetworkError	The connection failed after retries were exhausted.
TimeoutError	The request timed out after retries were exhausted.

Example

from korchestrator.remote import KorchestratorClient client = KorchestratorClient("https://engine.example.com", api_key="sk-example") client.get_run("r1") # doctest: +SKIP client.close()

get_run_summary

get_run_summary(
    run_id: str, *, timeout: float | None = None
) -> RunSummary

Fetch a run's summary (GET /v1/runs/{id}/summary).

Parameters:

Name	Type	Description	Default
run_id	str	The run to summarize.	required
timeout	float | None	Overrides the client's default timeout for this call only.	None

Returns:

Type	Description
RunSummary	The run's :class:~korchestrator.models.remote.RunSummary.

Raises:

Type	Description
ApiError	The engine rejected the request or the run does not exist.
NetworkError	The connection failed after retries were exhausted.
TimeoutError	The request timed out after retries were exhausted.

Example

from korchestrator.remote import KorchestratorClient client = KorchestratorClient("https://engine.example.com", api_key="sk-example") client.get_run_summary("r1") # doctest: +SKIP client.close()

list_keys

list_keys(
    *, timeout: float | None = None
) -> tuple[ApiKeySummary, ...]

List existing API keys (GET /v1/keys, the korchestrator:admin scope).

Parameters:

Name	Type	Description	Default
timeout	float | None	Overrides the client's default timeout for this call only.	None

Returns:

Type	Description
ApiKeySummary	The tenant's keys as :class:~korchestrator.models.remote.ApiKeySummary — never the
...	secret, which the engine only ever returns once, at creation.

Raises:

Type	Description
ApiError	The engine rejected the request (e.g. insufficient scope).
NetworkError	The connection failed after retries were exhausted.
TimeoutError	The request timed out after retries were exhausted.

Example

from korchestrator.remote import KorchestratorClient client = KorchestratorClient("https://engine.example.com", api_key="sk-example") client.list_keys() # doctest: +SKIP client.close()

list_runs

list_runs(
    *,
    tenant_id: str | None = None,
    timeout: float | None = None,
) -> tuple[RunSummary, ...]

List runs visible to the caller (GET /v1/runs).

Parameters:

Name	Type	Description	Default
tenant_id	str | None	Overrides the tenant derived from the API key, when authorized.	None
timeout	float | None	Overrides the client's default timeout for this call only.	None

Returns:

Type	Description
tuple[RunSummary, ...]	The visible runs as :class:~korchestrator.models.remote.RunSummary.

Raises:

Type	Description
ApiError	The engine rejected the request.
NetworkError	The connection failed after retries were exhausted.
TimeoutError	The request timed out after retries were exhausted.

Example

from korchestrator.remote import KorchestratorClient client = KorchestratorClient("https://engine.example.com", api_key="sk-example") client.list_runs() # doctest: +SKIP client.close()

me

me(*, timeout: float | None = None) -> CallerIdentity

Fetch the authenticated caller's identity (GET /v1/me).

Parameters:

Name	Type	Description	Default
timeout	float | None	Overrides the client's default timeout for this call only.	None

Returns:

Type	Description
CallerIdentity	The caller's :class:~korchestrator.models.remote.CallerIdentity.

Raises:

Type	Description
ApiError	The engine rejected the request.
NetworkError	The connection failed after retries were exhausted.
TimeoutError	The request timed out after retries were exhausted.

Example

from korchestrator.remote import KorchestratorClient client = KorchestratorClient("https://engine.example.com", api_key="sk-example") client.me() # doctest: +SKIP client.close()

models

models(
    *, timeout: float | None = None
) -> tuple[ModelCard, ...]

List the models available on the engine (GET /v1/models).

Parameters:

Name	Type	Description	Default
timeout	float | None	Overrides the client's default timeout for this call only.	None

Returns:

Type	Description
ModelCard	The available models as :class:~korchestrator.models.routing.ModelCard — the same
...	catalogue shape the local kernel's routing already uses.

Raises:

Type	Description
ApiError	The engine rejected the request.
NetworkError	The connection failed after retries were exhausted.
TimeoutError	The request timed out after retries were exhausted.

Example

from korchestrator.remote import KorchestratorClient client = KorchestratorClient("https://engine.example.com", api_key="sk-example") client.models() # doctest: +SKIP client.close()

my_quota

my_quota(*, timeout: float | None = None) -> Quota

Fetch the authenticated caller's usage quota (GET /v1/me/quota).

Parameters:

Name	Type	Description	Default
timeout	float | None	Overrides the client's default timeout for this call only.	None

Returns:

Type	Description
Quota	The caller's :class:~korchestrator.models.remote.Quota.

Raises:

Type	Description
ApiError	The engine rejected the request.
NetworkError	The connection failed after retries were exhausted.
TimeoutError	The request timed out after retries were exhausted.

Example

from korchestrator.remote import KorchestratorClient client = KorchestratorClient("https://engine.example.com", api_key="sk-example") client.my_quota() # doctest: +SKIP client.close()

my_runs

my_runs(
    *, timeout: float | None = None
) -> tuple[RunSummary, ...]

List the authenticated caller's runs (GET /v1/me/runs).

Parameters:

Name	Type	Description	Default
timeout	float | None	Overrides the client's default timeout for this call only.	None

Returns:

Type	Description
tuple[RunSummary, ...]	The caller's runs as :class:~korchestrator.models.remote.RunSummary.

Raises:

Type	Description
ApiError	The engine rejected the request.
NetworkError	The connection failed after retries were exhausted.
TimeoutError	The request timed out after retries were exhausted.

Example

from korchestrator.remote import KorchestratorClient client = KorchestratorClient("https://engine.example.com", api_key="sk-example") client.my_runs() # doctest: +SKIP client.close()

resume

resume(
    run_id: str, *, timeout: float | None = None
) -> RemoteRunResult

Resume a governance_paused run (POST /v1/run/{id}/resume).

Parameters:

Name	Type	Description	Default
run_id	str	The paused run to resume.	required
timeout	float | None	Overrides the client's default timeout for this call only.	None

Returns:

Type	Description
RemoteRunResult	The run's :class:~korchestrator.models.remote.RemoteRunResult after resuming.

Raises:

Type	Description
ApiError	The engine rejected the request or the run is not paused.
NetworkError	The connection failed after retries were exhausted.
TimeoutError	The request timed out after retries were exhausted.

Example

from korchestrator.remote import KorchestratorClient client = KorchestratorClient("https://engine.example.com", api_key="sk-example") client.resume("r1") # doctest: +SKIP client.close()

revoke_key

revoke_key(
    key_id: str, *, timeout: float | None = None
) -> None

Revoke an API key (DELETE /v1/keys/{id}, the korchestrator:admin scope).

Parameters:

Name	Type	Description	Default
key_id	str	The key to revoke.	required
timeout	float | None	Overrides the client's default timeout for this call only.	None

Raises:

Type	Description
ApiError	The engine rejected the request or the key does not exist.
NetworkError	The connection failed after retries were exhausted.
TimeoutError	The request timed out after retries were exhausted.

Example

from korchestrator.remote import KorchestratorClient client = KorchestratorClient("https://engine.example.com", api_key="sk-example") client.revoke_key("key-1") # doctest: +SKIP client.close()

run

run(
    objective: str,
    *,
    max_supersteps: int = 10,
    mock_mode: bool = False,
    tenant_id: str | None = None,
    timeout: float | None = None,
) -> RemoteRunResult

Start a run; the engine plans the graph (POST /v1/run/auto).

Returns as soon as the engine has accepted the run — typically status="started" or "running", not necessarily terminal. Use :meth:wait or :meth:run_and_wait to block until a terminal state.

Parameters:

Name	Type	Description	Default
objective	str	The goal, at least 10 characters.	required
max_supersteps	int	Hard halt bound for the run.	10
mock_mode	bool	Run with a deterministic mock model (spec 04 §7.1).	False
tenant_id	str | None	Overrides the tenant derived from the API key, when the caller is authorized to act cross-tenant.	None
timeout	float | None	Overrides the client's default timeout for this call only.	None

Returns:

Type	Description
RemoteRunResult	The run's initial :class:~korchestrator.models.remote.RemoteRunResult.

Raises:

Type	Description
ApiError	The engine rejected or failed the request.
NetworkError	The connection failed after retries were exhausted.
TimeoutError	The request timed out after retries were exhausted.

Example

from korchestrator.remote import KorchestratorClient client = KorchestratorClient("https://engine.example.com", api_key="sk-example") client.run("Summarize the quarterly incident reports") # doctest: +SKIP client.close()

run_and_wait

run_and_wait(
    objective: str,
    *,
    max_supersteps: int = 10,
    mock_mode: bool = False,
    tenant_id: str | None = None,
    poll_interval: float = DEFAULT_POLL_INTERVAL_SECONDS,
    timeout: float | None = None,
) -> RemoteRunResult

Start a run and block until it reaches a terminal state (run + wait).

Parameters:

Name	Type	Description	Default
objective	str	The goal, at least 10 characters.	required
max_supersteps	int	Hard halt bound for the run.	10
mock_mode	bool	Run with a deterministic mock model.	False
tenant_id	str | None	Overrides the tenant derived from the API key, when authorized.	None
poll_interval	float	Seconds between polls while waiting.	DEFAULT_POLL_INTERVAL_SECONDS
timeout	float | None	Per-request timeout passed to the start call and every poll.	None

Returns:

Type	Description
RemoteRunResult	The terminal :class:~korchestrator.models.remote.RemoteRunResult.

Raises:

Type	Description
ApiError	The engine rejected or failed the request.
NetworkError	The connection failed after retries were exhausted.
TimeoutError	A request timed out after retries were exhausted.

Example

from korchestrator.remote import KorchestratorClient client = KorchestratorClient("https://engine.example.com", api_key="sk-example") client.run_and_wait("Summarize Q3 incident reports") # doctest: +SKIP client.close()

run_swarm

run_swarm(
    agents: Sequence[AgentConfig],
    edges: Sequence[tuple[str, str]] = (),
    *,
    objective: str,
    max_supersteps: int = 10,
    tenant_id: str | None = None,
    timeout: float | None = None,
) -> RemoteRunResult

Start a run with an explicit graph (POST /v1/run/swarm).

Parameters:

Name	Type	Description	Default
agents	Sequence[AgentConfig]	The declared agents, as the same :class:AgentConfig shape Swarm uses.	required
edges	Sequence[tuple[str, str]]	(source_id, target_id) pairs; empty means every agent is independent.	()
objective	str	The goal, at least 10 characters.	required
max_supersteps	int	Hard halt bound for the run.	10
tenant_id	str | None	Overrides the tenant derived from the API key, when authorized.	None
timeout	float | None	Overrides the client's default timeout for this call only.	None

Returns:

Type	Description
RemoteRunResult	The run's initial :class:~korchestrator.models.remote.RemoteRunResult.

Raises:

Type	Description
ApiError	The engine rejected or failed the request.
NetworkError	The connection failed after retries were exhausted.
TimeoutError	The request timed out after retries were exhausted.

Example

from korchestrator.models import AgentConfig, AgentPersona from korchestrator.remote import KorchestratorClient client = KorchestratorClient("https://engine.example.com", api_key="sk-example") agent = AgentConfig(id="lead", persona=AgentPersona(role="lead")) client.run_swarm([agent], objective="Review this PR") # doctest: +SKIP client.close()

stream async

stream(
    run_id: str, *, timeout: float | None = None
) -> AsyncIterator[RunEvent]

Stream a run's live events (GET /v1/run/{id}/stream, Server-Sent Events).

Unlike every other method on this class, stream is a native async generator — SSE is inherently incremental, so wrapping it in :func:asyncio.run would defeat the point (spec 04 §7.5: "Streaming: SSE, exposed as an async iterator").

Reconnects automatically (up to max_retries consecutive failures, the same full-jitter backoff :meth:_request uses) if the connection drops before the engine closes the stream. The wire format (:class:~korchestrator.models.remote.RunEvent, one SSE frame per event) carries no event id, so a reconnect resumes from "now", not from the last event actually delivered — events emitted while disconnected are not replayed. The generator returns normally once the engine closes the stream (the run reached a terminal state).

Parameters:

Name	Type	Description	Default
run_id	str	The run to stream.	required
timeout	float | None	Overrides the client's default timeout for this call only.	None

Yields:

Name	Type	Description
Each	AsyncIterator[RunEvent]	class:~korchestrator.models.remote.RunEvent as the engine emits it.

Raises:

Type	Description
ApiError	The engine rejected the request or the run does not exist.
NetworkError	The connection failed after retries were exhausted.
TimeoutError	The connection timed out after retries were exhausted.

Example

import asyncio from korchestrator.remote import KorchestratorClient async def _demo(): ... client = KorchestratorClient( ... "https://engine.example.com", api_key="sk-example" ... ) ... async for event in client.stream("r1"): ... print(event.name) ... await client.aclose() asyncio.run(_demo()) # doctest: +SKIP

swarm_templates

swarm_templates(
    *, timeout: float | None = None
) -> tuple[SwarmTemplate, ...]

List the engine's swarm topology presets (GET /v1/swarm-templates).

Parameters:

Name	Type	Description	Default
timeout	float | None	Overrides the client's default timeout for this call only.	None

Returns:

Type	Description
tuple[SwarmTemplate, ...]	The available presets as :class:~korchestrator.models.remote.SwarmTemplate.

Raises:

Type	Description
ApiError	The engine rejected the request.
NetworkError	The connection failed after retries were exhausted.
TimeoutError	The request timed out after retries were exhausted.

Example

from korchestrator.remote import KorchestratorClient client = KorchestratorClient("https://engine.example.com", api_key="sk-example") client.swarm_templates() # doctest: +SKIP client.close()

tools

tools(
    *, timeout: float | None = None
) -> tuple[ToolDescriptor, ...]

List the tools available on the engine (GET /v1/tools).

Parameters:

Name	Type	Description	Default
timeout	float | None	Overrides the client's default timeout for this call only.	None

Returns:

Type	Description
tuple[ToolDescriptor, ...]	The available tools as :class:~korchestrator.models.remote.ToolDescriptor.

Raises:

Type	Description
ApiError	The engine rejected the request.
NetworkError	The connection failed after retries were exhausted.
TimeoutError	The request timed out after retries were exhausted.

Example

from korchestrator.remote import KorchestratorClient client = KorchestratorClient("https://engine.example.com", api_key="sk-example") client.tools() # doctest: +SKIP client.close()

wait

wait(
    run_id: str,
    *,
    poll_interval: float = DEFAULT_POLL_INTERVAL_SECONDS,
    timeout: float | None = None,
) -> RemoteRunResult

Block until run_id reaches a terminal state, polling :meth:get_run.

A governance_paused run is not terminal — it stays pending until an operator calls :meth:resume/cancel/edit_resume (P9.4), so this keeps polling through it rather than returning early.

Parameters:

Name	Type	Description	Default
run_id	str	The run to wait for.	required
poll_interval	float	Seconds between polls.	DEFAULT_POLL_INTERVAL_SECONDS
timeout	float | None	Per-request timeout passed to each poll; does not bound the total wait.	None

Returns:

Type	Description
RemoteRunResult	The terminal :class:~korchestrator.models.remote.RemoteRunResult.

Raises:

Type	Description
ApiError	The engine rejected the request or the run does not exist.
NetworkError	The connection failed after retries were exhausted.
TimeoutError	A poll request timed out after retries were exhausted.

Example

from korchestrator.remote import KorchestratorClient client = KorchestratorClient("https://engine.example.com", api_key="sk-example") client.wait("r1") # doctest: +SKIP client.close()

Wire-facing models

Distinct from the local kernel's own models — RemoteRunResult has no nested AgentState, for example, since that's an internal, non-serialised kernel detail the remote contract never exposes.

RemoteRunResult

Bases: BaseModel

A run's outcome as the remote engine reports it — terminal, paused, or still running.

Returned by :meth:KorchestratorClient.run, run_swarm, get_run, wait, and run_and_wait (spec 04 §7.3's POST /v1/run/auto, POST /v1/run/swarm, GET /v1/run/{id}).

RunSummary

Bases: BaseModel

A lightweight run summary — GET /v1/runs/{id}/summary and the terminal-state webhook.

Returned by :meth:KorchestratorClient.get_run_summary and, as a tuple, list_runs.

RunEvent

Bases: BaseModel

One live event from a run's SSE stream (GET /v1/run/{id}/stream, spec 04 §7.3/§7.5).

Deliberately mirrors :class:korchestrator.events.Event's shape (name/payload/ run_id) — the same concept, one streamed event — without importing across the clients/events module boundary spec 05's allowed-imports table doesn't authorize (clients/ reuses the same purpose-built-model pattern this file already uses for :class:RemoteRunResult rather than the local kernel's RunResult).

CallerIdentity

Bases: BaseModel

The authenticated caller's identity (GET /v1/me, spec 04 §7.3).

Quota

Bases: BaseModel

The caller's usage quota (GET /v1/me/quota, spec 04 §7.3).

ApiKey

Bases: BaseModel

A newly created API key (POST /v1/keys, spec 04 §7.3).

The secret is returned once, at creation — the engine never returns it again, matching every common API-key UX. key is a :class:~pydantic.SecretStr so an accidental repr/str (a log line, a stray print) never leaks it; call key.get_secret_value() to use it.

ApiKeySummary

Bases: BaseModel

An existing API key's metadata (GET /v1/keys, spec 04 §7.3) — never the secret.

ToolDescriptor

Bases: BaseModel

A discoverable tool (GET /v1/tools, spec 04 §7.3).

The wire twin of the local :class:korchestrator.interfaces.Connector protocol's name/description/schema (P6) — the same concept the AUB registry already exposes locally, described for a remote caller instead. The field is named input_schema, not schema — BaseModel reserves schema for its own (deprecated) .schema() method, and input_schema matches Anthropic's own tool-use vocabulary.

SwarmTemplate

Bases: BaseModel

A named swarm topology preset (GET /v1/swarm-templates, spec 04 §7.3).

Reuses :class:~korchestrator.models.agent.AgentConfig for agents — the same declarative shape run_swarm already accepts, so a template can be fetched and passed straight through.

ApiError

ApiError(
    message: str,
    *,
    status: int = 500,
    code: str | None = None,
    trace_id: str | None = None,
)

Bases: KorchError

A :class:~korchestrator.remote.KorchestratorClient call failed (spec 04 §7.5).

Raised for any non-2xx response the remote engine returns after retries are exhausted — status and trace_id let a caller correlate the failure with server-side logs and support without parsing the message text.

Parameters:

Name	Type	Description	Default
message	str	An actionable description of the failure.	required
status	int	The HTTP status code the engine returned. Defaults to 500 so ApiError stays constructible like every other KorchError when the status genuinely isn't known; real call sites always pass the response's actual status.	500
code	str | None	The stable error code — the engine's own code when its response body carries one, else :attr:default_code.	None
trace_id	str | None	The engine's request trace id, when the response provided one.	None

Example

from korchestrator.exceptions import ApiError, KorchError err = ApiError("Run 'r1' not found.", status=404, trace_id="trace-abc") err.status, err.trace_id, isinstance(err, KorchError) (404, 'trace-abc', True)

Build the error with the engine's HTTP status and optional trace id.