Python SDK

This guide shows how to install the Python SDK (enterprise-ai, import enterprise_ai), connect it to the Control Plane of the Navin Platform, and call every resource group of API v1 with typed errors, retries, pagination and streaming. Source: sdks/python/src/enterprise_ai. Every class, method and field is listed in the generated reference.

Version française : python.fr.md

Overview

The package gives two clients with the same surface:

Client	Execution model	When to use it
EnterpriseAI	Blocking calls over httpx.Client	Scripts, batch jobs, Django or Flask views, workers.
AsyncEnterpriseAI	Every method returns an awaitable, over httpx.AsyncClient	FastAPI, asyncio services, agents that run many calls at once.

Each client exposes one sub-client per resource group of API v1 (client.agents, client.gateway, client.finops...). A sub-client method maps one documented route, sends the authentication and tenancy headers, retries the safe failures and turns the error envelope into a typed exception. The same package also ships the enterprise-ai command line (CLI guide), the Navin Manager client enterprise_ai.manager (Navin Manager SDK) and offline validation of the v1 contracts (Contracts v1).

Requirements and compatibility

Item	Supported
Python	3.10, 3.11, 3.12 (requires-python = ">=3.10" in sdks/python/pyproject.toml)
Operating systems	Any OS where CPython runs: Linux (RHEL, Rocky, Alma 9, Ubuntu 22.04 and 24.04), macOS, Windows
Required dependencies	httpx>=0.25, pydantic>=2, jsonschema>=4.18, referencing>=0.30, pyyaml>=6
Control Plane API	v1 (enterprise_ai.API_VERSION == "v1")
SDK version	enterprise_ai.__version__ (1.0.2 for this release)

Optional extras install the dependencies of optional features only:

Extra	Installs	Needed for
otel	opentelemetry-api, opentelemetry-sdk	Reading the W3C traceparent of the current OpenTelemetry span.
realtime	websockets, livekit-api	client.realtime.connect(...) (voice live sessions).
langgraph	langgraph	The LangGraph runtime adapter of enterprise_ai.runtime.
temporal	temporalio	The Temporal runtime adapter of enterprise_ai.runtime.
dev	pytest, pytest-asyncio, ruff, build, websockets, langgraph	Working on the SDK itself.

Installation

1. Create a virtual environment. It isolates the SDK and its dependencies from the system Python.

   python3 -m venv .venv
   . .venv/bin/activate

   Expected result: python -c "import sys; print(sys.prefix)" prints the .venv folder.

2. Download the package from Harbor. The SDK packages are hosted in the sdk project of the Navinspire Harbor registry (artifact sdk/python, tag = SDK version). The Navinspire registry is registry.navinspire.ai; log in with a robot account that can pull from the sdk project (given by the platform team). If your organization mirrors the sdk project in its own Harbor, use the address of that mirror instead.

   oras login registry.navinspire.ai
   oras pull registry.navinspire.ai/sdk/python:1.0.2 -o sdk-python

   Expected result: the sdk-python folder holds the wheel enterprise_ai-1.0.2-py3-none-any.whl.

3. Install the package. Pick the extras your code needs.

   pip install sdk-python/enterprise_ai-1.0.2-py3-none-any.whl
   pip install "sdk-python/enterprise_ai-1.0.2-py3-none-any.whl[otel,realtime]"

   Expected result: python -c "import enterprise_ai; print(enterprise_ai.__version__)" prints 1.0.2, and enterprise-ai --help lists the CLI commands.

   If it fails: check that python3 --version is 3.10 or later, and the proxy settings of pip.

Offline install from a wheelhouse

Networks without Internet access install from a folder of wheels built once on a connected machine, from the wheel pulled from Harbor.

1. Build a wheelhouse on a connected machine and copy it. pip download fetches the wheel and every dependency for the target Python version; --no-index then forbids any network access during the install.

   pip download sdk-python/enterprise_ai-1.0.2-py3-none-any.whl -d wheelhouse
   pip install --no-index --find-links wheelhouse enterprise-ai==1.0.2

   Expected result: the install ends with Successfully installed enterprise-ai-1.0.2 without any download.

2. Or install from this repository. Used by the platform team and the CI.

   pip install -e sdks/python

Quickstart

This example lists the models the caller may use and sends one chat completion through the gateway, the only path to the models. Replace <token> with an API key of your tenant (see Authentication).

from enterprise_ai import EnterpriseAI

client = EnterpriseAI(base_url="<control-plane-url>", api_key="<token>", locale="en")
for model in client.models.list(limit=20):
    print(model["id"])
reply = client.gateway.chat("gpt-4o-mini", [{"role": "user", "content": "Bonjour"}])
print(reply["choices"][0]["message"]["content"])
client.close()

What happens: the client checks the base URL (https required except on localhost), sends the key as X-API-Key, iterates every page of GET /v1/models and posts the completion. Expected result: the model ids, then the text of the answer. A refused call raises a typed error such as PolicyDenied or QuotaExceeded (Errors).

Authentication

The tenant is always resolved by the server from the credential, never from the request body. The SDK supports these methods:

Method	Option	Environment variable	Header sent
Tenant API key eai_<prefix>_<secret> with scopes such as agents:run or models:invoke	api_key	EAI_API_KEY	X-API-Key
OIDC access token (Keycloak or the identity provider of the organization)	token	EAI_TOKEN	Authorization: Bearer <token>
Development mode, only on a server started with EAI_AUTH_DEV=1	tenant, dev_user, dev_roles	EAI_TENANT, EAI_DEV_USER, EAI_DEV_ROLES	X-Dev-Tenant, X-Dev-User (default sdk-dev), X-Dev-Roles
Workload identity (SPIFFE, mTLS) inside the mesh	none: the sidecar authenticates the workload	none	set by the mesh

When both api_key and token are set, both headers are sent. The development headers are sent only when neither an API key nor a token is configured. Create a key with client.identity.create_api_key(name, scopes) (the secret is shown once) and rotate it with client.identity.rotate_api_key(key_id).

import os

from enterprise_ai import EnterpriseAI

os.environ["EAI_BASE_URL"] = "<control-plane-url>"
os.environ["EAI_API_KEY"] = "<token>"
client = EnterpriseAI()  # every setting not given comes from the EAI_* variables
key = client.identity.create_api_key("ci-pipeline", ["agents:run"])

danger

Never ship an API key in a browser or mobile application, and never write it in a source file: read it from the environment or a secret store.

Client configuration

EnterpriseAI(base_url=None, *, http_client=None, transport=None, **settings) and AsyncEnterpriseAI(...) accept every field of ClientConfig as a keyword argument. A setting left out falls back to its environment variable, then to its default.

Name	Type	Default	Environment variable	Description
base_url	str | None	http://127.0.0.1:8000	EAI_BASE_URL	Control Plane root without /v1. https is required except for localhost, 127.0.0.1 and ::1; a URL with embedded credentials is refused.
api_key	str | None	None	EAI_API_KEY	Tenant API key, sent as X-API-Key.
token	str | None	None	EAI_TOKEN	OIDC access token, sent as Authorization: Bearer.
tenant	str | None	None	EAI_TENANT	Tenant slug, sent as X-Dev-Tenant only without credential (development mode).
project	str | None	None	EAI_PROJECT	Project slug, sent as X-Project.
environment	str | None	None	EAI_ENVIRONMENT	dev, test, staging or prod, sent as X-Environment.
locale	str | None	en	EAI_LOCALE	en or fr (fr-FR gives fr); sent as Accept-Language, selects the language of error messages.
timeout	float	30.0	none	Timeout of one HTTP attempt, in seconds.
retries	int	3	none	Retries after the first attempt (Retries and timeouts).
backoff	float	0.5	none	First backoff delay in seconds, doubled at each attempt, plus up to 50 % jitter.
max_backoff	float	20.0	none	Upper bound of one backoff delay and of an honored Retry-After, in seconds.
dev_user	str | None	None	EAI_DEV_USER	Development user, sent as X-Dev-User.
dev_roles	str | None	None	EAI_DEV_ROLES	Development roles (comma separated), sent as X-Dev-Roles.
traceparent	str | None	None	none	Fixed W3C traceparent for every call (the current span wins when one is active).
headers	dict[str, str]	{}	none	Extra headers sent with every request; they override the defaults.
ca_bundle	str | None	None	EAI_CA_BUNDLE	PEM file of extra trusted CAs; verification stays on.
dangerously_disable_tls_verification_for_localhost	bool	False	none	Disables certificate checks; refused unless the base URL host is local.

Constructor-only arguments:

Name	Type	Default	Description
http_client	httpx.Client (httpx.AsyncClient for the async client)	None	A client you built yourself (proxy, mTLS, connection pool).
transport	httpx.BaseTransport (httpx.AsyncBaseTransport)	None	A transport used to build the client, for example httpx.MockTransport in tests.
other keyword arguments	any	none	Passed to httpx.Client (for example proxy= or limits=); verify and follow_redirects=True are refused.

repr(client.config) masks the API key and the token (eai_***). Redirects are never followed, so credentials never reach another origin; TLS 1.2 is the minimum version.

Requests and responses

A method takes the path parameters as positional arguments and the body fields as keyword arguments; None values are dropped and an extra= or **extra argument merges further fields. It returns the decoded JSON body: a dict or a list, typed by the v1 contracts and the OpenAPI document. A list route returns a Page (Pagination); a 204 returns None; a text route such as client.system.metrics() returns str.

agent = client.agents.get("agt_ap")
print(agent["id"], agent.get("status"))
response = client.gateway.chat("gpt-4o-mini", [{"role": "user", "content": "Hi"}], raw=True)
print(response.status, response.request_id, response.decision_id, response.cost_estimated, response.cost_measured)

raw=True (on gateway.chat) returns an ApiResponse with data, status, headers and the properties request_id (X-Request-Id), decision_id (X-EAI-Decision-Id), cost_estimated and cost_measured (X-EAI-Cost-Estimated, X-EAI-Cost-Measured). The Pydantic models of the contracts are validated offline with enterprise_ai.contracts.validate (Contracts v1).

Errors

Every error answer uses the envelope {"error": {"code", "message", "message_i18n": {"en", "fr"}, "details", "request_id"}}. The SDK raises the most specific subclass of EaiError, chosen by the code first, then by the HTTP status. Every error has code, status, message (in the client locale), messages (message_i18n), details (secret looking keys replaced by ***) and request_id.

Class	Codes	HTTP status	Retried by the SDK
EaiError	any other code	any other status	no
AuthError	unauthenticated, unauthorized, forbidden, invalid_credentials	401, 403	no
PolicyDenied	policy_denied	403	no
QuotaExceeded	quota_exceeded, budget_exceeded	429	yes, as a 429, until the retries are spent
EffortNotSupported	effort_not_supported	422	no
ContextOverflow	context_overflow	422	no
ApprovalRequired	approval_required (details.approval_id)	409	no
NotFound	not_found	404	no
ValidationError	validation_error, invalid_request, schema_invalid	400, 422	no
ConflictError	conflict	409	no
RateLimited	rate_limited	429	yes, then raised
ServerError	server_error, authz_backend_error	5xx	yes for safe calls (500, 502, 503, 504), then raised
TransportError	transport_error	none (no answer)	yes for safe calls, then raised
EgressDenied	egress_denied (.reason)	403	no
PromptInjectionBlocked	prompt_injection_blocked	403	no
PayloadTooLarge	payload_too_large	413	no
AuthzUnconfirmed	authz_unconfirmed	none (raised by the SDK)	no
BudgetExhausted	budget_exhausted	none (local agent loop)	no
AgentStopped	agent_stopped	none (kill switch seen by the agent)	no

from enterprise_ai import ApprovalRequired, PolicyDenied, QuotaExceeded

try:
    client.gateway.chat("gpt-4o-mini", [{"role": "user", "content": "Bonjour"}])
except QuotaExceeded as err:
    print(err.code, err.message, err.messages.get("fr"), err.request_id)
except PolicyDenied as err:
    print(err.details.get("reasons"), err.details.get("decision_id"))
except ApprovalRequired as err:
    client.approvals.get(err.details["approval_id"])

str(err) gives [<code>] <message> (request_id=<id>): quote the request id to the platform team, it links the call to the audit trail and the traces.

Retries and timeouts

The transport retries a failed attempt up to retries times (default 3) with an exponential backoff (backoff * 2^attempt plus up to 50 % jitter, at most max_backoff). A Retry-After header (seconds or HTTP date) replaces the computed delay, bounded by max_backoff.

Failure	GET, HEAD, OPTIONS, PUT, DELETE	POST, PATCH with an Idempotency-Key	POST, PATCH without key
HTTP 429	retried	retried	retried
HTTP 500, 502, 503, 504	retried	retried	not retried
Network error (no answer)	retried	retried	not retried
Any other status (400, 401, 403, 404, 409, 413, 422)	not retried	not retried	not retried

timeout (default 30 s) applies to each attempt. Disable retries for one client with retries=0.

from enterprise_ai import EnterpriseAI

batch = EnterpriseAI(base_url="<control-plane-url>", api_key="<token>", timeout=120.0, retries=5,
                     backoff=1.0, max_backoff=30.0)

Pagination

List routes take limit and cursor and answer {"items": [...], "next_cursor": "..."}. A list method returns a Page: page.items (the first page), page.next_cursor, page.raw (the decoded body), page.next_page() (the next Page or None), len(page) and page[i]. Iterating the page follows every cursor until the end.

first = client.runs.list(agent_id="agt_ap", limit=50)
print(len(first), first.next_cursor)
for run in first:  # every page, one item at a time
    print(run["id"], run.get("status"))
second = first.next_page()

Filters are keyword arguments (state=, agent_id=...), limit= and cursor= are common to every list method, and params={...} passes a filter the method does not name. The async client returns an AsyncPage, iterated with async for.

Idempotency and long operations

The calls that create something with a cost or a side effect send an Idempotency-Key header: a UUID generated per call and kept across the retries of that call, or the idempotency_key= you pass. The Control Plane replays the first answer for a key it already saw, so a retry never creates a duplicate run, reservation or approval. The methods are agents.run, runs.spawn_subagent, a2a.send, approvals.create, tools.invoke, quotas.reserve, quotas.commit, quotas.release, finops.ingest_usage, finops.ingest_usage_batch, finops.import_invoice, finops.create_media_price, rag.ingest, sandbox.execute, realtime.create_session, realtime.record_consent, multimodal.video.generate, evaluation.run, evaluation.redteam, privacy.erase_subject, audit.export, channels.inbound, applications.create, applications.deploy, delegations.create, mandates.create, break_glass.create, evidence.create_hold, license.activate, models.create_version, marketplace.publish_package, marketplace.import_publication, marketplace.install, soc.create_detection, soc.create_order, soc.create_recovery, storage.create_profile, organizations.create and business_units.create.

run = client.agents.run("agt_ap", {"invoice_media_id": "med_01HZX3K9"}, idempotency_key="invoice-2291")
again = client.agents.run("agt_ap", {"invoice_media_id": "med_01HZX3K9"}, idempotency_key="invoice-2291")

Long work is asynchronous on the server. Runs, sandbox executions, multimodal jobs and approvals are read again until they reach a final state; helpers wait for you:

execution = client.sandbox.execute("python:3.12-slim", ["python", "-c", "print(1 + 1)"], profile="gvisor")
done = client.sandbox.wait(execution["id"], timeout=120)
job = client.multimodal.job("job_1")
operation = client.system.operation("op_1")  # GET /v1/operations/{operation_id}

client.approvals.wait(approval_id, timeout=...) and client.sandbox.wait(...) poll until the state is final (wait_async on the async client). client.system.operation(operation_id) reads a long operation announced by the Operation-Id header of a long creation.

Async and concurrency

AsyncEnterpriseAI has the same sub-clients and methods; each returns an awaitable. Use it as an async context manager so the connection pool is closed.

import asyncio

from enterprise_ai import AsyncEnterpriseAI


async def main() -> None:
    async with AsyncEnterpriseAI(base_url="<control-plane-url>", api_key="<token>") as aclient:
        models, me = await asyncio.gather(aclient.models.list(), aclient.identity.me())
        async for run in await aclient.runs.list(limit=50):
            print(run["id"])
        async with aclient.quotas.areservation(cost="0.50", tokens=4000, scope={"agent": "agt_ap"}) as r:
            out = await aclient.gateway.chat("gpt-4o-mini", [{"role": "user", "content": "Bonjour"}])
            r.commit(tokens=out["usage"]["total_tokens"])


asyncio.run(main())

One EnterpriseAI instance can be shared by the threads of a process: it holds no per call state, and httpx.Client is thread safe. One AsyncEnterpriseAI instance belongs to one event loop.

Streaming and realtime

client.gateway.chat(..., stream=True) returns a ChatStream over Server-Sent Events: iterate it for the chat.completion.chunk dicts, call text() for the text deltas only, or final() for the merged completion with its usage. client.runs.events(run_id) streams the events of a run; client.telemetry.stream() streams telemetry.

stream = client.gateway.chat("gpt-4o-mini", [{"role": "user", "content": "Bonjour"}], stream=True)
for delta in stream.text():
    print(delta, end="")
completion = stream.accumulator.completion()
for event in client.runs.events("run_1"):
    print(event.event, event.json())

Voice live sessions (client.realtime.create_session, client.realtime.connect) are described in the Realtime guide.

Logging and debugging

The SDK writes no log of its own. The HTTP layer logs through the standard logging module: the httpx logger writes one INFO line per request (method, URL, status), and httpcore writes the connection details at DEBUG. Secrets are never in these lines: the credentials travel in headers, which are not logged.

import logging

logging.basicConfig(level=logging.INFO)
logging.getLogger("httpx").setLevel(logging.INFO)

When a call fails, print the error (str(err) holds the code and the request id) and give the request_id to the platform team: it is the key of the audit event and of the trace. print(client.config) shows the effective settings with the secrets masked.

Testing

Inject an httpx.MockTransport: no server, no network, and the real SDK code runs (headers, errors, pagination).

import httpx

from enterprise_ai import EnterpriseAI, NotFound


def handler(request: httpx.Request) -> httpx.Response:
    if request.url.path == "/v1/agents/missing":
        return httpx.Response(404, json={"error": {"code": "not_found", "message": "Resource not found."}})
    return httpx.Response(200, json={"items": [{"id": "agt_1"}], "next_cursor": None})


test_client = EnterpriseAI(base_url="<control-plane-url>", api_key="eai_test_key",
                           transport=httpx.MockTransport(handler))
assert [a["id"] for a in test_client.agents.list()] == ["agt_1"]
try:
    test_client.agents.get("missing")
except NotFound as err:
    assert err.code == "not_found"

For the async client pass an httpx.MockTransport with an async handler to AsyncEnterpriseAI(transport=...). Contract documents are checked without a server with enterprise_ai.contracts.validate.

Versioning and compatibility

• The SDK follows semantic versioning: 1.x keeps every public name of enterprise_ai; a removal or an incompatible change waits for 2.0.
• It targets API v1 and the contracts v1 (Contracts v1); fields added by the server are kept in the returned dicts, so a newer Control Plane does not break an older SDK.
• enterprise_ai.__version__ and the User-Agent: enterprise-ai-python/<version> header identify the SDK in the audit trail.
• Pin the minor version in production (enterprise-ai~=1.0) and read the release notes before an upgrade.

Resource guides

Each subsection lists the sub-clients of one resource group and one example per common call. The parameters, types and defaults of every method are in the generated reference.

System, contracts and manifests

client.system reads the platform state; client.contracts validates documents against the v1 schemas; client.manifests applies GitOps manifests.

meta = client.system.meta()
ready = client.system.readyz()
contracts = client.contracts.list()
check = client.contracts.validate("agent-manifest", {"apiVersion": "eai/v1", "kind": "Agent"})

Identity, tenants and access

client.identity (principal, members, API keys, service identities, relationship tuples, secrets), client.tenants, client.projects, client.organizations, client.business_units and client.authz.

me = client.identity.me()
member = client.identity.add_member("<user-email>", "developer")
project = client.projects.create("invoices", "Invoices")
client.projects.create_environment("prj_1", "staging")
allowed = client.authz.check("user:alice", "viewer", "agent:agt_ap")

Policies, risk, incidents and approvals

client.policies (policy contracts, evaluation, decision log, revisions, bundles, simulation), client.risk, client.incidents, client.governance and client.approvals (separation of duties: the requester cannot approve).

decision = client.policies.evaluate({"subject": {"id": "usr_alice", "kind": "user"},
                                     "resource": {"type": "model", "id": "gpt-4o-mini"},
                                     "action": "model.invoke", "context": {}})
approval = client.approvals.create("tool.invoke", "tool", resource_id="erp.pay", reason="payment run")
client.approvals.decide("apr_3", "approve", comment="checked")
incident = client.incidents.create("Prompt injection on agt_ap", "high")

Models, providers, routes and the gateway

client.providers, client.models (registry, Capability Resolver, versions and rollouts), client.routes (Model Router), client.context (context budgets), client.gateway (chat, embeddings, rerank, images, audio, video, estimates), client.llm (named interface), client.llm_management and client.isolation.

plan = client.models.resolve("o3-mini", effort="high")
client.providers.test("prv_openai")
simulation = client.routes.simulate(model="default-chat", data_classification="confidential")
vectors = client.gateway.embeddings("text-embedding-3-small", ["invoice", "receipt"])
estimate = client.gateway.estimate("gpt-4o-mini", [{"role": "user", "content": "Bonjour"}], max_output_tokens=500)
budget = client.context.plan(128000, 4000, segments=[{"kind": "rag", "tokens": 90000}])

gateway.chat(model, messages, effort=...) first checks the effort against the model registry (precheck=True): an effort the model cannot honor raises EffortNotSupported before any call, and a mapped effort emits an EffortAdjustedWarning. Pass precheck=False to leave the decision to the server.

Agents and runs

client.agents (registry, versions, suspend, resume, kill, run), client.runs (state, steps, replay, events, checkpoints, heartbeats, sub-agents), client.sessions, client.a2a, client.schedules, client.triggers and client.sandbox.

agent = client.agents.create({"apiVersion": "eai/v1", "kind": "Agent",
                              "metadata": {"name": "ap", "version": "1.0.0", "owner": "finance"}, "spec": {}})
run = client.agents.run("agt_ap", {"invoice_media_id": "med_01HZX3K9"})
client.runs.heartbeat("run_1", progress=0.4)
client.runs.checkpoint("run_1", {"page": 3}, label="after-ocr")
child = client.runs.spawn_subagent("run_1", "agt_ocr", task={"media_id": "med_01HZX3K9"})
client.schedules.create("agt_ap", "0 6 * * 1-5", input={})
client.agents.suspend("agt_ap", reason="incident INC-8")

Tools, MCP, skills, plugins, connectors and channels

client.tools (registry and Tool Gateway), client.mcp (servers and JSON-RPC gateway), client.mcp_runtime, client.skills, client.plugins, client.connectors (write-only credentials, leases, webhooks), client.channels, client.navin_library and client.file_security.

result = client.tools.invoke("erp.lookup", {"invoice": "INV-2291"}, run_id="run_1")
tools = client.mcp.server_tools("mcp_files")
answer = client.mcp.rpc("mcp_files", "tools/call", {"name": "read", "arguments": {"path": "/a.txt"}})
client.skills.verify("invoice-reader")
client.connectors.set_credentials("sharepoint", {"client_secret": "<secret>"})
client.channels.inbound("web", "Bonjour", user="u1")

Webhooks pushed to the platform carry X-EAI-Tenant, X-EAI-Delivery (deduplication) and X-EAI-Signature: t=<unix>,v1=<hex HMAC-SHA256 of "<t>.<raw body>">. verify_webhook compares in constant time and refuses a tampered body, a wrong secret, a malformed header or a timestamp older than tolerance seconds.

from enterprise_ai.connectors import sign_webhook, verify_webhook

raw_body = b'{"event": "invoice"}'
headers = sign_webhook("<secret>", raw_body, tenant="acme", delivery_id="evt-42")
assert verify_webhook("<secret>", raw_body, headers["X-EAI-Signature"], tolerance=300)

Data: privacy, DLP, encryption, RAG, memory and prompts

client.privacy, client.dlp, client.tokenization, client.encryption, client.rag (knowledge bases with ACL aware retrieval), client.memory and client.prompts.

client.dlp.redact("My IBAN is FR76 3000 6000 0112 3456 7890 189", mode="mask")
tokens = client.tokenization.tokenize(values=["FR7630006000011234567890189"])
sealed = client.encryption.encrypt(b"secret", "tenant-default")
kb = client.rag.create("policies")
client.rag.ingest("kb_1", content="Payment terms are 30 days.", title="handbook")
hits = client.rag.query("kb_1", "payment terms", top_k=5)
client.memory.write("Preferred supplier is ACME", scope={"type": "agent", "id": "agt_ap"})
text = client.prompts.render("invoice-extract", {"lang": "fr"}, version="3")

Multimodal and realtime

client.multimodal (media upload and jobs) with vision, image, video, audio, stt and tts, and client.realtime (voice live sessions, consents, turns, transcripts).

media = client.multimodal.upload(b"\x89PNG", "scan.png", content_type="image/png")
client.multimodal.vision.analyze("Total amount?", media_id="med_1", model="vision-default")
client.multimodal.stt.transcribe("med_AUD1", language="fr")
client.multimodal.tts.synthesize("Bonjour", voice="fr_FR-siwis-medium")
session = client.realtime.create_session(pipeline="stt", config={"language": "fr"})

Quality, audit and observability

client.evaluation (suites, runs, quality gates, red team, comparisons), client.audit (events, chain verification, WORM export, correlations) and client.telemetry (spans, overview, traces, SLO).

gate = client.evaluation.check_gate({"type": "agent", "ref": "agt_ap", "version": "1.4.0"})
chain = client.audit.verify()
for event in client.audit.events(action="agent.apply", limit=100):
    print(event["id"])
slo = client.telemetry.slo(window_minutes=60)

FinOps and quotas

client.finops (estimates, usage events, summaries, price cards, budgets, reconciliation) and client.quotas (quotas and reservations). A reservation holds budget before a call and is committed with the measured cost.

client.finops.estimate("local", model_id="piper", usage_type="tts", units={"characters": 1200})
summary = client.finops.summary(group_by="agent")
with client.quotas.reservation(cost="0.50", tokens=4000, scope={"agent": "agt_ap"}) as reservation:
    out = client.gateway.chat("gpt-4o-mini", [{"role": "user", "content": "Bonjour"}])
    reservation.commit(tokens=out["usage"]["total_tokens"])

Security and licence

client.security (posture, settings, incidents, run cancellation) and client.license (licence state, activation, history).

posture = client.security.posture()
client.security.update_settings(prompt_injection_mode="block")
licence = client.license.get()

Governance and platform resources

Sub-clients added for the licence, delegation, SOC, storage, marketplace, evidence and application routes of API v1. Every snippet below is executed by sdks/python/tests/test_governance_clients.py against an httpx.MockTransport. List routes with a cursor (delegations.list, license.history) return a Page.

Attribute	Paths
license	/v1/license, /v1/license/activations, /v1/license/history
execution_context	/v1/context/issue, /v1/context/verify, /v1/context/{id}/revoke, /v1/context/jwks, /v1/context/keys
delegations, mandates, break_glass	/v1/delegations, /v1/mandates, /v1/break-glass
soc	/v1/soc/detections, /v1/soc/orders, /v1/soc/kill-switches, /v1/soc/receipts, /v1/soc/recoveries, /v1/soc/transactions, /v1/soc/incidents/{id}/evidence
storage	/v1/storage/profiles (verify, checks, presign)
marketplace	/v1/marketplace/publishers, /v1/marketplace/packages, /v1/marketplace/preflight, /v1/marketplace/installations
evidence	/v1/evidence/verify, /v1/evidence/seals, /v1/evidence/holds, /v1/evidence/retention, /v1/evidence/runs/{id}
applications	/v1/applications (deployments, status, external agents)
isolation	/v1/llm/isolation
models	/v1/models/{id}/versions, /v1/models/{id}/rollout, /v1/models/{id}/rollback
finops, evaluation, policies, identity, rag, mcp, realtime	media prices and costs, comparisons, bundles, revisions, engines, simulation, whoami, quarantine and review, catalogue and discovery, consents, turns, transcripts

Realtime WebSocket helpers: create_response(text, voice=None) (response.create), cancel_response() (response.cancel), speech_started() (input_audio.speech_started, barge-in), commit_audio() (input_audio.commit); server events response.created, response.done, response.cancelled and recording.saved are listed in SERVER_EVENTS.

client.license.get()
client.license.activate("eyJhbGciOiJFZERTQSJ9.licence")
entries = list(client.license.history(limit=50))

ctx = client.execution_context.issue(audience="tool-gateway", ttl_seconds=300)
client.execution_context.verify("eyJ.context", audience="tool-gateway")
d = client.delegations.create("agent:billing", scopes=["invoices:read"], ttl_seconds=600)
client.delegations.authorize("invoices.read", delegation_id=d["id"])
client.mandates.create("agt_billing", "monthly close", "svc_billing", scopes=["invoices:read"])
grant = client.break_glass.create("production outage INC-42", 900, roles=["tenant_admin"])
client.break_glass.revoke(grant["id"], reason="resolved")

client.soc.create_detection("prompt_injection", severity="high", run_id="run_1")
order = client.soc.create_order("suspend_agent", {"agent": "agt_1"}, "injection confirmed")
client.soc.approve_order(order["id"])
client.soc.verify_kill_switch("ks_1")
recovery = client.soc.create_recovery("ks_1", "root cause fixed")
client.soc.approve_recovery(recovery["id"])

profile = client.storage.create_profile("evidence", "s3_compatible", "eai-evidence",
                                       endpoint="<s3-endpoint>", credential_ref="openbao://storage/evidence")
client.storage.verify_profile(profile["id"], probe_write=True)
client.storage.presign(profile["id"], "reports/q1.pdf", "read", expires_seconds=300)

client.marketplace.preflight("hr-pack", "1.2.0", "staging")
inst = client.marketplace.install("hr-pack", "1.2.0", "staging", approved_permissions=["hr.leave.read"])
client.marketplace.apply_installation(inst["id"])
client.marketplace.rollback_installation(inst["id"])
# kill switch: suspend one capability now, resume only with an approval of two other people
client.marketplace.set_installation_status(inst["id"], "suspended", permissions=["hr.leave.read"], reason="incident 7")
# 202 with an approval of two other people, then the same call with approval_id=...
client.marketplace.set_installation_status(inst["id"], "installed")
client.marketplace.set_package_status("hr-pack", "suspended", reason="CVE")
client.marketplace.supervision(inst["id"])

client.evidence.verify()
seal = client.evidence.create_seal()
client.evidence.verify_seal(seal["id"])
client.evidence.create_hold("case-42", "litigation")
client.evidence.run_retention(kind="retention", dry_run=True)

app = client.applications.create("HR assistant", "assistant", risk_level="medium")
dep = client.applications.deploy(app["id"], "1.0.0", "staging")
client.applications.activate(app["id"], dep["id"])
client.applications.set_status(app["id"], "suspended", "incident INC-42")

client.models.create_version("claude-sonnet-eu", {"version": "2026-01"})
client.models.start_rollout("claude-sonnet-eu", "2026-01", {"percent": 10})
client.models.rollout("claude-sonnet-eu")
client.models.rollback("claude-sonnet-eu")
client.isolation.get(project_id="prj_1")

client.finops.estimate_media_cost({"unit": "image", "quantity": 4})
client.evaluation.compare({"suite_id": "st_1", "baseline_run_id": "er_1", "candidate_run_id": "er_2"})
client.policies.simulate({"rules": []}, policy_id="pol_1", recorded=True)
client.policies.diff("pol_1", from_revision=1, to_revision=2)
client.identity.whoami()
client.rag.review("kb_1", "doc_1", "release", reason="false positive")
client.mcp.discover("mcp_1")
client.realtime.record_consent("rt_1", "support call", 30, transcription_allowed=True, method="verbal")
client.realtime.turns("rt_1")

Platform Manager space of the console (ADR-0128, platform_admin only): client.platform_manager relays every call to the Navin Platform Manager on behalf of the caller, whose RBAC, approvals and stop rules decide; a Manager refusal is raised with its status and a platform_manager_<code> code.

client.platform_manager.summary()                  # services total, healthy, warning, failed per cluster
client.platform_manager.services("c1")             # five state fields and administrative state
client.platform_manager.request_operation("c1", "opa", "rolling_restart", batch_size=1)
client.platform_manager.backups()                  # backup artefacts with sha256 and last verification (ADR-0162)
client.platform_manager.plan_restore("b1", "isolated")  # restore plan, two approvers other than the requester
client.platform_manager.dr_exercises()             # DR exercises with the observed RPO and RTO
client.platform_manager.console()                  # Manager console and operator console links

Navin Manager client

The Navin Manager (navin-manager) installs and operates the infrastructure: hosts, clusters, services, high availability, backups, restores and disaster recovery. Its Python client is enterprise_ai.manager.ManagerClient (and AsyncManagerClient), configured with a manager API token nmt_<12 hex>_<secret> and the NM_MANAGER_* variables, never with the EAI_* credentials. The full guide, with every group and the CLI, is Navin Manager SDK.

from enterprise_ai.manager import ManagerConfig

config = ManagerConfig.from_env({"NM_MANAGER_URL": "<manager-url>",
                                 "NM_MANAGER_TOKEN_FILE": "/run/secrets/manager.token"})
print(config)  # the token is masked

client.platform_manager is a different path: the Platform Manager space of the console, relayed by the Control Plane for a platform_admin.

Reference

• Generated Python reference: every class, method, parameter and field.
• API v1 of the Control Plane and Contracts v1.
• CLI guide and CLI reference.
• Other SDKs.