SDK Python

Ce guide montre comment installer le SDK Python (enterprise-ai, import enterprise_ai), le connecter au Control Plane de la Navin Platform et appeler chaque groupe de ressources de l'API v1 avec des erreurs typées, des reprises, la pagination et le streaming. Source : sdks/python/src/enterprise_ai. Chaque classe, méthode et champ figure dans la référence générée.

English version: python.md

Vue d'ensemble

Le paquet fournit deux clients avec la même surface :

Client	Modèle d'exécution	Quand l'utiliser
EnterpriseAI	Appels bloquants sur httpx.Client	Scripts, traitements par lots, vues Django ou Flask, workers.
AsyncEnterpriseAI	Chaque méthode renvoie un awaitable, sur httpx.AsyncClient	FastAPI, services asyncio, agents qui lancent beaucoup d'appels à la fois.

Chaque client expose un sous-client par groupe de ressources de l'API v1 (client.agents, client.gateway, client.finops...). Une méthode de sous-client correspond à une route documentée : elle envoie les en-têtes d'authentification et de tenancy, reprend les échecs sans risque et transforme l'enveloppe d'erreur en exception typée. Le même paquet livre aussi la ligne de commande enterprise-ai (guide de la CLI), le client du Navin Manager enterprise_ai.manager (SDK du Navin Manager) et la validation hors ligne des contrats v1 (Contrats v1).

Prérequis et compatibilité

Élément	Pris en charge
Python	3.10, 3.11, 3.12 (requires-python = ">=3.10" dans sdks/python/pyproject.toml)
Systèmes d'exploitation	Tout système où CPython fonctionne : Linux (RHEL, Rocky, Alma 9, Ubuntu 22.04 et 24.04), macOS, Windows
Dépendances requises	httpx>=0.25, pydantic>=2, jsonschema>=4.18, referencing>=0.30, pyyaml>=6
API du Control Plane	v1 (enterprise_ai.API_VERSION == "v1")
Version du SDK	enterprise_ai.__version__ (1.0.2 pour cette version)

Les extras optionnels n'installent que les dépendances des fonctions optionnelles :

Extra	Installe	Nécessaire pour
otel	opentelemetry-api, opentelemetry-sdk	Lire le traceparent W3C du span OpenTelemetry courant.
realtime	websockets, livekit-api	client.realtime.connect(...) (sessions vocales en direct).
langgraph	langgraph	L'adaptateur d'exécution LangGraph de enterprise_ai.runtime.
temporal	temporalio	L'adaptateur d'exécution Temporal de enterprise_ai.runtime.
dev	pytest, pytest-asyncio, ruff, build, websockets, langgraph	Travailler sur le SDK lui-même.

Installation

1. Créez un environnement virtuel. Il isole le SDK et ses dépendances du Python du système.

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

   Résultat attendu : python -c "import sys; print(sys.prefix)" affiche le dossier .venv.

2. Téléchargez le paquet depuis Harbor. Les paquets des SDK sont hébergés dans le projet sdk du registre Harbor de Navinspire (artefact sdk/python, tag = version du SDK). Le registre Navinspire est registry.navinspire.ai ; connectez-vous avec un compte robot qui peut lire (pull) le projet sdk (fourni par l'équipe plateforme). Si votre organisation réplique le projet sdk dans son propre Harbor, utilisez l'adresse de ce miroir.

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

   Résultat attendu : le dossier sdk-python contient la wheel enterprise_ai-1.0.2-py3-none-any.whl.

3. Installez le paquet. Choisissez les extras dont votre code a besoin.

   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]"

   Résultat attendu : python -c "import enterprise_ai; print(enterprise_ai.__version__)" affiche 1.0.2, et enterprise-ai --help liste les commandes de la CLI.

   En cas d'échec : vérifiez que python3 --version est 3.10 ou plus, et les réglages de proxy de pip.

Installation hors ligne depuis un dépôt de wheels

Les réseaux sans accès à Internet installent depuis un dossier de wheels construit une fois sur une machine connectée, à partir de la wheel récupérée dans Harbor.

1. Préparez un dépôt de wheels sur une machine connectée et copiez-le. pip download récupère la wheel et chaque dépendance pour la version de Python cible ; --no-index interdit ensuite tout accès réseau pendant l'installation.

   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

   Résultat attendu : l'installation se termine par Successfully installed enterprise-ai-1.0.2 sans aucun téléchargement.

2. Ou installez depuis ce dépôt. Utilisé par l'équipe plateforme et la CI.

   pip install -e sdks/python

Démarrage rapide

Cet exemple liste les modèles que l'appelant peut utiliser et envoie une complétion de chat par le gateway, seul chemin vers les modèles. Remplacez <token> par une clé API de votre tenant (voir Authentification).

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()

Ce qui se passe : le client vérifie l'URL de base (https obligatoire sauf sur localhost), envoie la clé dans X-API-Key, parcourt chaque page de GET /v1/models et poste la complétion. Résultat attendu : les identifiants des modèles, puis le texte de la réponse. Un appel refusé lève une erreur typée comme PolicyDenied ou QuotaExceeded (Erreurs).

Authentification

Le tenant est toujours résolu par le serveur à partir de l'identifiant, jamais à partir du corps de la requête. Le SDK prend en charge ces méthodes :

Méthode	Option	Variable d'environnement	En-tête envoyé
Clé API de tenant eai_<prefix>_<secret> avec des scopes comme agents:run ou models:invoke	api_key	EAI_API_KEY	X-API-Key
Jeton d'accès OIDC (Keycloak ou le fournisseur d'identité de l'organisation)	token	EAI_TOKEN	Authorization: Bearer <token>
Mode développement, uniquement sur un serveur lancé avec EAI_AUTH_DEV=1	tenant, dev_user, dev_roles	EAI_TENANT, EAI_DEV_USER, EAI_DEV_ROLES	X-Dev-Tenant, X-Dev-User (défaut sdk-dev), X-Dev-Roles
Identité de workload (SPIFFE, mTLS) dans le maillage	aucune : le sidecar authentifie le workload	aucune	posé par le maillage

Quand api_key et token sont tous deux renseignés, les deux en-têtes sont envoyés. Les en-têtes de développement ne partent que si ni clé API ni jeton n'est configuré. Créez une clé avec client.identity.create_api_key(name, scopes) (le secret n'est affiché qu'une fois) et faites-la tourner avec 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()  # chaque réglage non donné vient des variables EAI_*
key = client.identity.create_api_key("ci-pipeline", ["agents:run"])

danger

Ne livrez jamais une clé API dans une application navigateur ou mobile, et ne l'écrivez jamais dans un fichier source : lisez-la depuis l'environnement ou un coffre de secrets.

Configuration du client

EnterpriseAI(base_url=None, *, http_client=None, transport=None, **settings) et AsyncEnterpriseAI(...) acceptent chaque champ de ClientConfig comme argument nommé. Un réglage absent prend la valeur de sa variable d'environnement, puis sa valeur par défaut.

Nom	Type	Défaut	Variable d'environnement	Description
base_url	str | None	http://127.0.0.1:8000	EAI_BASE_URL	Racine du Control Plane sans /v1. https est obligatoire sauf pour localhost, 127.0.0.1 et ::1 ; une URL avec des identifiants intégrés est refusée.
api_key	str | None	None	EAI_API_KEY	Clé API du tenant, envoyée dans X-API-Key.
token	str | None	None	EAI_TOKEN	Jeton d'accès OIDC, envoyé dans Authorization: Bearer.
tenant	str | None	None	EAI_TENANT	Slug du tenant, envoyé dans X-Dev-Tenant seulement sans identifiant (mode développement).
project	str | None	None	EAI_PROJECT	Slug du projet, envoyé dans X-Project.
environment	str | None	None	EAI_ENVIRONMENT	dev, test, staging ou prod, envoyé dans X-Environment.
locale	str | None	en	EAI_LOCALE	en ou fr (fr-FR donne fr) ; envoyé dans Accept-Language, choisit la langue des messages d'erreur.
timeout	float	30.0	aucune	Délai d'une tentative HTTP, en secondes.
retries	int	3	aucune	Reprises après la première tentative (Reprises et délais).
backoff	float	0.5	aucune	Premier délai d'attente en secondes, doublé à chaque tentative, plus jusqu'à 50 % d'aléa.
max_backoff	float	20.0	aucune	Borne d'un délai d'attente et d'un Retry-After respecté, en secondes.
dev_user	str | None	None	EAI_DEV_USER	Utilisateur de développement, envoyé dans X-Dev-User.
dev_roles	str | None	None	EAI_DEV_ROLES	Rôles de développement (séparés par des virgules), envoyés dans X-Dev-Roles.
traceparent	str | None	None	aucune	traceparent W3C fixe pour chaque appel (le span courant l'emporte quand il existe).
headers	dict[str, str]	{}	aucune	En-têtes supplémentaires envoyés avec chaque requête ; ils remplacent les valeurs par défaut.
ca_bundle	str | None	None	EAI_CA_BUNDLE	Fichier PEM d'autorités de certification supplémentaires ; la vérification reste active.
dangerously_disable_tls_verification_for_localhost	bool	False	aucune	Désactive le contrôle des certificats ; refusé sauf si l'hôte de l'URL de base est local.

Arguments réservés au constructeur :

Nom	Type	Défaut	Description
http_client	httpx.Client (httpx.AsyncClient pour le client asynchrone)	None	Un client construit par vos soins (proxy, mTLS, pool de connexions).
transport	httpx.BaseTransport (httpx.AsyncBaseTransport)	None	Un transport utilisé pour construire le client, par exemple httpx.MockTransport dans les tests.
autres arguments nommés	quelconque	aucun	Transmis à httpx.Client (par exemple proxy= ou limits=) ; verify et follow_redirects=True sont refusés.

repr(client.config) masque la clé API et le jeton (eai_***). Les redirections ne sont jamais suivies, donc les identifiants n'atteignent jamais une autre origine ; TLS 1.2 est la version minimale.

Requêtes et réponses

Une méthode prend les paramètres de chemin en arguments positionnels et les champs du corps en arguments nommés ; les valeurs None sont retirées et un argument extra= ou **extra fusionne d'autres champs. Elle renvoie le corps JSON décodé : un dict ou une list, typés par les contrats v1 et le document OpenAPI. Une route de liste renvoie une Page (Pagination) ; un 204 renvoie None ; une route texte comme client.system.metrics() renvoie un 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 (sur gateway.chat) renvoie une ApiResponse avec data, status, headers et les propriétés request_id (X-Request-Id), decision_id (X-EAI-Decision-Id), cost_estimated et cost_measured (X-EAI-Cost-Estimated, X-EAI-Cost-Measured). Les modèles Pydantic des contrats se valident hors ligne avec enterprise_ai.contracts.validate (Contrats v1).

Erreurs

Chaque réponse d'erreur utilise l'enveloppe {"error": {"code", "message", "message_i18n": {"en", "fr"}, "details", "request_id"}}. Le SDK lève la sous-classe la plus précise de EaiError, choisie d'abord par le code, puis par le statut HTTP. Chaque erreur porte code, status, message (dans la langue du client), messages (message_i18n), details (clés d'allure secrète remplacées par ***) et request_id.

Classe	Codes	Statut HTTP	Reprise par le SDK
EaiError	tout autre code	tout autre statut	non
AuthError	unauthenticated, unauthorized, forbidden, invalid_credentials	401, 403	non
PolicyDenied	policy_denied	403	non
QuotaExceeded	quota_exceeded, budget_exceeded	429	oui, comme un 429, jusqu'à épuisement des reprises
EffortNotSupported	effort_not_supported	422	non
ContextOverflow	context_overflow	422	non
ApprovalRequired	approval_required (details.approval_id)	409	non
NotFound	not_found	404	non
ValidationError	validation_error, invalid_request, schema_invalid	400, 422	non
ConflictError	conflict	409	non
RateLimited	rate_limited	429	oui, puis levée
ServerError	server_error, authz_backend_error	5xx	oui pour les appels sans risque (500, 502, 503, 504), puis levée
TransportError	transport_error	aucun (pas de réponse)	oui pour les appels sans risque, puis levée
EgressDenied	egress_denied (.reason)	403	non
PromptInjectionBlocked	prompt_injection_blocked	403	non
PayloadTooLarge	payload_too_large	413	non
AuthzUnconfirmed	authz_unconfirmed	aucun (levée par le SDK)	non
BudgetExhausted	budget_exhausted	aucun (boucle d'agent locale)	non
AgentStopped	agent_stopped	aucun (kill switch vu par l'agent)	non

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) donne [<code>] <message> (request_id=<id>) : citez l'identifiant de requête à l'équipe plateforme, il relie l'appel à la piste d'audit et aux traces.

Reprises et délais

Le transport reprend une tentative en échec jusqu'à retries fois (3 par défaut) avec un délai exponentiel (backoff * 2^tentative plus jusqu'à 50 % d'aléa, au plus max_backoff). Un en-tête Retry-After (secondes ou date HTTP) remplace le délai calculé, borné par max_backoff.

Échec	GET, HEAD, OPTIONS, PUT, DELETE	POST, PATCH avec une Idempotency-Key	POST, PATCH sans clé
HTTP 429	repris	repris	repris
HTTP 500, 502, 503, 504	repris	repris	non repris
Erreur réseau (pas de réponse)	repris	repris	non repris
Tout autre statut (400, 401, 403, 404, 409, 413, 422)	non repris	non repris	non repris

timeout (30 s par défaut) s'applique à chaque tentative. Désactivez les reprises d'un client avec 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

Les routes de liste prennent limit et cursor et répondent {"items": [...], "next_cursor": "..."}. Une méthode de liste renvoie une Page : page.items (la première page), page.next_cursor, page.raw (le corps décodé), page.next_page() (la Page suivante ou None), len(page) et page[i]. Parcourir la page suit chaque curseur jusqu'à la fin.

first = client.runs.list(agent_id="agt_ap", limit=50)
print(len(first), first.next_cursor)
for run in first:  # toutes les pages, un élément à la fois
    print(run["id"], run.get("status"))
second = first.next_page()

Les filtres sont des arguments nommés (state=, agent_id=...), limit= et cursor= sont communs à toutes les méthodes de liste, et params={...} transmet un filtre que la méthode ne nomme pas. Le client asynchrone renvoie une AsyncPage, parcourue avec async for.

Idempotence et opérations longues

Les appels qui créent quelque chose avec un coût ou un effet de bord envoient un en-tête Idempotency-Key : un UUID généré par appel et conservé entre les reprises de cet appel, ou la valeur idempotency_key= que vous passez. Le Control Plane rejoue la première réponse pour une clé déjà vue, donc une reprise ne crée jamais un run, une réservation ou une approbation en double. Les méthodes sont 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 et 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")

Le travail long est asynchrone côté serveur. Les runs, les exécutions en sandbox, les tâches multimodales et les approbations se relisent jusqu'à un état final ; des aides attendent pour vous :

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=...) et client.sandbox.wait(...) interrogent jusqu'à un état final (wait_async sur le client asynchrone). client.system.operation(operation_id) lit une opération longue annoncée par l'en-tête Operation-Id d'une création longue.

Asynchrone et concurrence

AsyncEnterpriseAI a les mêmes sous-clients et méthodes ; chacune renvoie un awaitable. Utilisez-le comme gestionnaire de contexte asynchrone pour que le pool de connexions soit fermé.

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())

Une instance de EnterpriseAI peut être partagée par les threads d'un processus : elle ne garde aucun état par appel, et httpx.Client est sûr entre threads. Une instance de AsyncEnterpriseAI appartient à une seule boucle d'événements.

Streaming et temps réel

client.gateway.chat(..., stream=True) renvoie un ChatStream sur Server-Sent Events : parcourez-le pour obtenir les dicts chat.completion.chunk, appelez text() pour les seuls deltas de texte, ou final() pour la complétion fusionnée avec sa consommation. client.runs.events(run_id) diffuse les événements d'un run ; client.telemetry.stream() diffuse la télémétrie.

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())

Les sessions vocales en direct (client.realtime.create_session, client.realtime.connect) sont décrites dans le guide temps réel.

Journalisation et débogage

Le SDK n'écrit aucun journal propre. La couche HTTP journalise par le module standard logging : le logger httpx écrit une ligne INFO par requête (méthode, URL, statut), et httpcore écrit le détail des connexions en DEBUG. Ces lignes ne contiennent jamais de secret : les identifiants voyagent dans des en-têtes, qui ne sont pas journalisés.

import logging

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

Quand un appel échoue, affichez l'erreur (str(err) contient le code et l'identifiant de requête) et donnez le request_id à l'équipe plateforme : c'est la clé de l'événement d'audit et de la trace. print(client.config) montre les réglages effectifs avec les secrets masqués.

Tests

Injectez un httpx.MockTransport : pas de serveur, pas de réseau, et le vrai code du SDK s'exécute (en-têtes, erreurs, 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"

Pour le client asynchrone, passez un httpx.MockTransport avec un gestionnaire asynchrone à AsyncEnterpriseAI(transport=...). Les documents de contrat se vérifient sans serveur avec enterprise_ai.contracts.validate.

Versions et compatibilité

• Le SDK suit le versionnage sémantique : 1.x garde chaque nom public de enterprise_ai ; une suppression ou un changement incompatible attend 2.0.
• Il cible l'API v1 et les contrats v1 (Contrats v1) ; les champs ajoutés par le serveur restent dans les dicts renvoyés, donc un Control Plane plus récent ne casse pas un SDK plus ancien.
• enterprise_ai.__version__ et l'en-tête User-Agent: enterprise-ai-python/<version> identifient le SDK dans la piste d'audit.
• Fixez la version mineure en production (enterprise-ai~=1.0) et lisez les notes de version avant une mise à jour.

Guides des ressources

Chaque sous-section liste les sous-clients d'un groupe de ressources et un exemple par appel courant. Les paramètres, types et défauts de chaque méthode sont dans la référence générée.

Système, contrats et manifestes

client.system lit l'état de la plateforme ; client.contracts valide des documents contre les schémas v1 ; client.manifests applique des manifestes GitOps.

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

Identité, tenants et accès

client.identity (principal, membres, clés API, identités de service, tuples de relation, secrets), client.tenants, client.projects, client.organizations, client.business_units et 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")

Politiques, risques, incidents et approbations

client.policies (contrats de politique, évaluation, journal des décisions, révisions, bundles, simulation), client.risk, client.incidents, client.governance et client.approvals (séparation des tâches : le demandeur ne peut pas approuver).

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")

Modèles, fournisseurs, routes et gateway

client.providers, client.models (registre, Capability Resolver, versions et déploiements progressifs), client.routes (Model Router), client.context (budgets de contexte), client.gateway (chat, embeddings, rerank, images, audio, vidéo, estimations), client.llm (interface nommée), client.llm_management et 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=...) vérifie d'abord l'effort dans le registre des modèles (precheck=True) : un effort que le modèle ne sait pas honorer lève EffortNotSupported avant tout appel, et un effort converti émet un EffortAdjustedWarning. Passez precheck=False pour laisser la décision au serveur.

Agents et runs

client.agents (registre, versions, suspension, reprise, arrêt, lancement), client.runs (état, étapes, rejeu, événements, points de reprise, heartbeats, sous-agents), client.sessions, client.a2a, client.schedules, client.triggers et 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")

Outils, MCP, skills, plugins, connecteurs et canaux

client.tools (registre et Tool Gateway), client.mcp (serveurs et gateway JSON-RPC), client.mcp_runtime, client.skills, client.plugins, client.connectors (identifiants en écriture seule, baux, webhooks), client.channels, client.navin_library et 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")

Les webhooks poussés vers la plateforme portent X-EAI-Tenant, X-EAI-Delivery (dédoublonnage) et X-EAI-Signature: t=<unix>,v1=<hex HMAC-SHA256 de "<t>.<corps brut>">. verify_webhook compare en temps constant et refuse un corps modifié, un mauvais secret, un en-tête mal formé ou un horodatage plus ancien que tolerance secondes.

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)

Données : confidentialité, DLP, chiffrement, RAG, mémoire et prompts

client.privacy, client.dlp, client.tokenization, client.encryption, client.rag (bases de connaissances avec recherche respectant les ACL), client.memory et 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 et temps réel

client.multimodal (dépôt de médias et tâches) avec vision, image, video, audio, stt et tts, et client.realtime (sessions vocales en direct, consentements, tours de parole, transcriptions).

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"})

Qualité, audit et observabilité

client.evaluation (suites, exécutions, quality gates, red team, comparaisons), client.audit (événements, vérification de la chaîne, export WORM, corrélations) et client.telemetry (spans, vue d'ensemble, 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 et quotas

client.finops (estimations, événements d'usage, synthèses, grilles de prix, budgets, rapprochement) et client.quotas (quotas et réservations). Une réservation retient du budget avant un appel et se confirme avec le coût mesuré.

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"])

Sécurité et licence

client.security (posture, réglages, incidents, annulation de run) et client.license (état de la licence, activation, historique).

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

Ressources de gouvernance et de plateforme

Sous-clients ajoutés pour les routes de licence, de délégation, de SOC, de stockage, de place de marché, de preuves et d'applications de l'API v1. Chaque extrait ci-dessous est exécuté par sdks/python/tests/test_governance_clients.py contre un httpx.MockTransport. Les listes à curseur (delegations.list, license.history) renvoient une Page.

Attribut	Chemins
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 (vérification, historique, URL présignées)
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 (déploiements, statut, agents externes)
isolation	/v1/llm/isolation
models	/v1/models/{id}/versions, /v1/models/{id}/rollout, /v1/models/{id}/rollback
finops, evaluation, policies, identity, rag, mcp, realtime	prix et coûts média, comparaisons, bundles, révisions, moteurs, simulation, whoami, quarantaine et revue, catalogue et découverte, consentements, tours de parole, transcriptions

Aides WebSocket temps réel : create_response(text, voice=None) (response.create), cancel_response() (response.cancel), speech_started() (input_audio.speech_started, interruption), commit_audio() (input_audio.commit) ; les événements serveur response.created, response.done, response.cancelled et recording.saved figurent dans 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 : suspendre une capacité tout de suite, reprendre seulement avec l'approbation de deux autres personnes
client.marketplace.set_installation_status(inst["id"], "suspended", permissions=["hr.leave.read"], reason="incident 7")
# 202 avec une approbation de deux autres personnes, puis le même appel avec 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")

Espace Platform Manager de la console (ADR-0128, réservé à platform_admin) : client.platform_manager relaie chaque appel au Navin Platform Manager au nom de l'appelant, dont le RBAC, les approbations et les règles d'arrêt décident ; un refus du Manager est levé avec son statut et un code platform_manager_<code>.

client.platform_manager.summary()                  # services au total, sains, en alerte, en échec par cluster
client.platform_manager.services("c1")             # cinq champs d'état et état administratif
client.platform_manager.request_operation("c1", "opa", "rolling_restart", batch_size=1)
client.platform_manager.backups()                  # artefacts de sauvegarde, sha256 et dernière vérification (ADR-0162)
client.platform_manager.plan_restore("b1", "isolated")  # plan de restauration, deux approbateurs autres que le demandeur
client.platform_manager.dr_exercises()             # exercices de PRA avec RPO et RTO observés
client.platform_manager.console()                  # liens vers la console du Manager et les consoles d'exploitation

Client du Navin Manager

Le Navin Manager (navin-manager) installe et exploite l'infrastructure : hôtes, clusters, services, haute disponibilité, sauvegardes, restaurations et reprise après sinistre. Son client Python est enterprise_ai.manager.ManagerClient (et AsyncManagerClient), configuré avec un jeton d'API du manager nmt_<12 hex>_<secret> et les variables NM_MANAGER_*, jamais avec les identifiants EAI_*. Le guide complet, avec chaque groupe et la CLI, est SDK du Navin Manager.

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)  # le jeton est masqué

client.platform_manager est un autre chemin : l'espace Platform Manager de la console, relayé par le Control Plane pour un platform_admin.

Référence

• Référence Python générée : chaque classe, méthode, paramètre et champ.
• API v1 du Control Plane et Contrats v1.
• Guide de la CLI et référence de la CLI.
• Autres SDK.