Intégrer NVIDIA NeMo Fabric via le SDK Python

Utilisez cette skill quand une base de code consommatrice — une application, un service, un harness d'évaluation ou une plateforme — doit exécuter des harnesses d'agent via le SDK Python typé de NeMo Fabric. Le consommateur possède son propre objet de configuration et le traduit en FabricConfig en mémoire ; NeMo Fabric gère la sélection des adaptateurs, le cycle de vie du runtime et les résultats normalisés.

Limite d'intégration

Utilisez le contrat public en mémoire. Ces règles maintiennent une intégration consommatrice supportée et sûre pour les mises à niveau :

Importez uniquement depuis le package public nemo_fabric. N'importez jamais _native ni aucun module interne à un adaptateur.
Construisez la configuration comme une FabricConfig typée en mémoire et passez-la directement à NeMo Fabric. Créez chaque variante de déploiement ou d'évaluation avec des fonctions Python ordinaires et model_copy(deep=True). Une intégration de plateforme peut sérialiser la config typée à l'intérieur d'une spécification de lancement transient privée quand elle franchit une limite de processus ; ce transport n'est pas un format d'authoring public.
Laissez NeMo Fabric contrôler le harness. Ne réimplémentez pas la logique de démarrage, d'invocation ou d'arrêt, et ne gérez pas directement les threads, sessions ou processus des adaptateurs.
Traitez runtime_id, invocation_id et request_id comme des chaînes de corrélation opaques, non analysables ou réutilisables.

Consultez config-mapping.md pour savoir comment traduire un objet de configuration consommatrice en FabricConfig, et pour la liste complète des mécanismes qui restent cachés derrière cette limite.

Installer et configurer l'environnement

Le consommateur ou son environnement d'exécution possède l'installation ; NeMo Fabric valide les hypothèses du runtime mais n'installe jamais de harnesses ou d'identifiants au moment du lancement.

NeMo Fabric supporte Python 3.11 à 3.14. Utilisez Python 3.11 à 3.13 pour Hermes Agent ; l'intégration Harbor nécessite Python 3.12 ou ultérieur.
Installez le runtime avec uv pip install nemo-fabric (ajoutez l'extra harbor pour l'intégration Harbor). Consultez le guide d'installation.
Sélectionnez l'adaptateur harness via HarnessConfig.adapter_id. Pour installer le runtime NeMo Fabric, l'adaptateur et le harness supporté dans un même environnement, utilisez nemo-fabric[claude], nemo-fabric[codex] ou nemo-fabric[deepagents].
Hermes Agent 0.20 et ultérieur n'est plus installable depuis PyPI. Suivez le guide d'installation d'Hermes Agent, puis installez le package nemo-fabric[hermes-agent] dans l'environnement Python qui exécute Hermes Agent. Ces packages n'installent pas Hermes Agent.
Dans un environnement d'adaptateur séparé, installez nemo-fabric-adapters-<adapter>[harness]. Ceci installe l'adaptateur et les dépendances du harness supporté sans le runtime NeMo Fabric. Utilisez full à la place quand ce package d'adaptateur propose des intégrations optionnelles installables via package.
Pointez le runtime vers un environnement d'adaptateur séparé avec ADAPTER_PYTHON. Utilisez les versions de release NeMo Fabric correspondantes pour le runtime et le package adaptateur à moins qu'un appairage différent n'ait été explicitement validé.
Si l'environnement d'adaptateur gère déjà un harness compatible, installez la distribution nemo-fabric-adapters-<adapter> simple. Les distributions d'adaptateur simples contiennent uniquement les dépendances du runtime détenues par l'adaptateur.
Les packages d'adaptateurs LangChain Deep Agents et Hermes Agent fournissent relay et incluent le package Python NeMo Relay dans full. Les extras Hermes Agent n'installent pas Hermes Agent. Claude et Codex ne fournissent pas relay ; leurs extras harness et full installent le CLI nemo-relay supporté aux côtés du SDK harness.
Fournissez les identifiants du modèle via des variables d'environnement nommées par la configuration (ModelConfig.api_key_env), jamais comme littéraux dans le code.
Confirmez que l'extension native est importable ; les appels SDK lèvent FabricNativeUnavailableError quand elle est manquante.

Construire la config typée à partir de la config consommatrice

Mappez l'objet d'application, de travail ou de déploiement du consommateur dans une FabricConfig avec les modèles publics et les méthodes d'assistance :

from nemo_fabric import (
    FabricConfig,
    HarnessConfig,
    InstructionConfig,
    InstructionsConfig,
    MetadataConfig,
    ModelConfig,
    RuntimeConfig,
    ToolsConfig,
)


def to_tools_config(job) -> ToolsConfig | None:
    enabled = job.enabled_tools
    blocked = list(job.blocked_tools)
    if enabled is None and not blocked:
        return None
    return ToolsConfig(
        enabled=None if enabled is None else list(enabled),
        blocked=blocked,
    )


def to_fabric_config(job) -> FabricConfig:
    config = FabricConfig(
        metadata=MetadataConfig(name=job.name),
        harness=HarnessConfig(adapter_id=job.adapter_id, resolution="preinstalled"),
        models={
            "default": ModelConfig(
                provider=job.provider,
                model=job.model,
                api_key_env=job.api_key_env,
                base_url=job.base_url,
            )
        },
        instructions=(
            InstructionsConfig(
                system=InstructionConfig(content=job.system_instruction),
            )
            if job.system_instruction is not None
            else None
        ),
        runtime=RuntimeConfig(
            input_schema="chat",
            output_schema="message",
            timeout_seconds=job.timeout_seconds,
            max_turns=job.max_turns,
        ),
        tools=to_tools_config(job),
    )
    config.add_skill_path(job.skill_dir)
    config.add_mcp_server(
        "github",
        transport="streamable-http",
        url="${GITHUB_MCP_URL}",
        exposure="harness_native",
    )
    return config

Structurez les capacités avec ToolsConfig, add_tool_definition, block_tools, add_skill_path, remove_skill_path, add_mcp_server, remove_mcp_server et enable_relay.
Utilisez add_tool_definition uniquement quand l'adaptateur sélectionné accepte tools.definitions et publie un tool_definition_schema.
Utilisez une liste allowed_tools restreinte ou un blocked_tools non vide sur add_mcp_server uniquement quand l'adaptateur sélectionné déclare à la fois mcp et mcp.tool_filters. Un serveur non filtré nécessite uniquement mcp. allowed_tools=None expose chaque outil découvert, tandis qu'une liste vide n'en expose aucun ; les outils bloqués sont supprimés après application de cette liste blanche. Les noms d'outils doivent être non vides, et la planification rejette un outil qui apparaît dans les deux listes.
Configurez l'authentification MCP uniquement quand l'adaptateur sélectionné déclare mcp.auth.oauth2 ou mcp.auth.service_account, en correspondance avec le type d'authentification.
Créez des variantes de déploiement ou d'évaluation avec model_copy(deep=True) et des fonctions Python ordinaires ; chaque copie se planifie et s'exécute indépendamment.
Passez base_dir=... à tout appel Fabric quand la config utilise des chemins relatifs, afin que les skills, espaces de travail et artefacts s'ancrent à la mise en page du consommateur.

Le dépôt exemple code_review_agent montre ce modèle de bout en bout avec des variantes complètes d'Hermes Agent, Codex, Deep Agents, d'environnement, MCP et de télémétrie. Réutilisez-le plutôt que de dupliquer la construction de configuration.

Choisir un cycle de vie

Choisissez le plus petit cycle de vie dont le consommateur a besoin :

Invocation unique — une entrée, aucun état retenu après l'appel. await Fabric().run(config, input=...) exécute le cycle complet de démarrage, invocation et arrêt et retourne un RunResult. Passez request=RunRequest(...) au lieu de input=... quand l'invocation a besoin d'un ID de requête détenu par l'appelant ou d'un contexte (les deux s'excluent mutuellement).
Runtime avec état — tours ordonnés sur un cycle de vie logique d'un harness. Démarrez-le avec start_runtime(...) et utilisez le Runtime retourné comme gestionnaire de contexte async afin que le nettoyage s'exécute à la sortie — l'arrêt est tenté, non garanti (stop() peut lever FabricRuntimeError ; voir Consommer les résultats et gérer les erreurs). Un runtime accepte une invocation active à la fois ; les appels qui se chevauchent lèvent FabricStateError.
Stream OpenAI natif — chunks OpenAI Chat Completions natifs de l'adaptateur plus un résultat normalisé final séparé. Vérifiez runtime.supports_openai_streaming, appelez runtime.invoke_openai_stream(...), itérez le OpenAIInvokeStream retourné, puis attendez stream.result(). Le descripteur d'adaptateur sélectionné doit déclarer capabilities.streaming. Chaque mapping cédé a object == "chat.completion.chunk" ; un stream vide est valide. Si l'itération s'arrête tôt, appelez await stream.aclose() pour vider sans annuler l'invocation cible. Ce chemin ne nécessite pas NeMo Relay ni streaming=True.
Stream NVIDIA NeMo Relay — enregistrements ATOF bruts en direct plus un résultat normalisé final. Activez NeMo Relay, passez streaming=True à start_runtime(...), appelez runtime.invoke_stream(...), itérez le InvokeStream retourné, puis attendez stream.result(). La fin de l'itération n'indique pas le succès de l'invocation ; les exceptions d'invocation sont levées de result(), tandis que les échecs rapportés par le harness restent des valeurs RunResult normalisées. Si l'itération s'arrête tôt, appelez await stream.aclose() avant de commencer un autre tour. aclose() attend que le tour se termine ; il n'annule pas l'invocation du harness. Le SDK expose intentionnellement uniquement les enregistrements ATOF générés par NeMo Relay. Ce chemin est indépendant du streaming OpenAI natif. L'écouteur limite chaque enregistrement à 1 Mio et sa file d'attente à 1 024 enregistrements ou 16 Mio de données encodées. Il corrige les enregistrements via l'ID de requête NeMo Fabric pour les harnesses en processus. Pour les harnesses gateway, il utilise le rôle de portée de tour NeMo Relay et l'index de tour basé à 1. Il cède uniquement l'arborescence de portée correspondante. Les enregistrements de tour antérieur retardés n'entrent donc pas dans le prochain stream. Si les séquences de tour gateway et NeMo Fabric ne s'alignent pas, le SDK rejette les enregistrements non corrélés et émet un RuntimeWarning après l'épuisement naturel du stream. L'écouteur se lie à NEMO_FABRIC_STREAMING_HOST, qui est par défaut 127.0.0.1. Surchargez-le quand la gateway doit atteindre le SDK via une autre interface réseau, et restreignez l'accès à cette interface. Si l'itération async atteint son délai de drainage post-tour sans connexion NeMo Relay, ou reçoit des données sans une racine de tour correspondante, le SDK émet un RuntimeWarning pour ce mode de défaillance ; les appelants qui attendent uniquement stream.result() n'exécutent pas cette vérification d'avertissement. Le SDK avertit aussi quand un téléchargement NeMo Relay se termine avant de compléter son corps de requête par chunks parce que les enregistrements cédés peuvent être incomplets. L'indicateur streaming=True n'active pas NeMo Relay par lui-même. Sans streaming=True, le démarrage laisse la configuration NeMo Relay inchangée et n'injecte pas le récepteur de stream ATOF détenu par le SDK.

L'adaptateur sélectionné possède la topologie d'exécution. Les adaptateurs Claude, Codex, Deep Agents et Hermes Agent regroupés conservent leur client natif, graphe/checkpointer ou agent/base de données à l'intérieur d'un seul hôte local pour l'ensemble du runtime. Les adaptateurs locaux process et python utilisent ce cycle de vie d'hôte ; les consommateurs ne sélectionnent pas un autre mécanisme d'exécution local dans FabricConfig. Ne rejouez pas une invocation après une défaillance du runtime. Arrêtez le runtime défaillant et démarrez explicitement un nouveau selon la politique de tentative de l'application.

Le fragment de cycle de vie ci-dessous montre les formes disponibles. Il suppose que l'appelant a déjà défini config = to_fabric_config(job) et choisi base, comme décrit dans l'exemple de configuration ci-dessus :

import asyncio

from nemo_fabric import Fabric


async def main() -> None:
    fabric = Fabric()

    # Single invocation
    result = await fabric.run(config, base_dir=base, input="Review the changes.")

    # Multi-turn
    async with await fabric.start_runtime(config, base_dir=base) as runtime:
        first = await runtime.invoke(input="Inspect the repository")
        second = await runtime.invoke(input="Now review the latest patch")

    # Adapter-native OpenAI Chat Completions chunks
    async with await fabric.start_runtime(config, base_dir=base) as runtime:
        if runtime.supports_openai_streaming:
            stream = runtime.invoke_openai_stream(input="Review the latest patch")
            async for chunk in stream:
                print(chunk)
            openai_streamed_result = await stream.result()

    # NeMo Relay streaming
    streaming_config = config.model_copy(deep=True).enable_relay()
    async with await fabric.start_runtime(
        streaming_config,
        base_dir=base,
        streaming=True,
    ) as runtime:
        stream = runtime.invoke_stream(input="Review the latest patch")
        async for record in stream:
            print(record)
        streamed_result = await stream.result()


asyncio.run(main())

NeMo Fabric ne possède aucune file d'attente de planification d'application, groupe de workers, politique de tentative ou politique de concurrence globale. Chaque runtime ne permet toujours qu'une invocation active ; démarrez des runtimes indépendants pour le travail parallèle. Le chemin de streaming NeMo Relay utilise une file d'attente de transport bornée interne et la contre-pression TCP uniquement pour transporter les enregistrements ATOF d'une invocation. Traitez stream.result() comme faisant autorité, et reconstruisez le travail imbriqué à partir des champs ATOF uuid et parent_uuid plutôt que de l'ordre du stream.

Pour le streaming OpenAI natif, le SDK possède le transport HTTP loopback authentifié, l'encadrement NDJSON par chunks et les valeurs de corrélation. Le code consommateur n'en fournit pas. L'adaptateur exécute exactement une invocation, et le RunResult terminal reste séparé du stream de chunks. Consommez entièrement le stream ou appelez await stream.aclose() avant de commencer un autre tour. Attendre stream.result() draine et jette aussi les chunks OpenAI natifs non lus, alors consommez l'itérateur d'abord quand l'application a besoin de chaque chunk.

Valider avant d'exécuter

Résolvez et diagnostiquez avant de dépenser de l'effort sur un runtime, particulièrement dans un nouvel environnement ou avant de dépendre d'une capacité optionnelle :

fabric = Fabric()
plan = fabric.plan(config, base_dir=base)             # sync: adapter + capabilities
report = await fabric.doctor(config, base_dir=base)   # async: preflight checks

print(plan.adapter.adapter_id, report.status)

Utilisez plan(...) pour confirmer la sélection d'adaptateur et le routage de capacités avant d'exécuter. La planification valide harness.settings par rapport au descripteur d'adaptateur exactement résolu et, quand présent, workflow.settings par rapport au descripteur de cible d'adaptateur exactement résolu.
Utilisez doctor(...) pour vérifier la disponibilité de l'adaptateur, la résolution, le contexte d'environnement et les exigences déclarées telles que les variables d'environnement requises. Son status agrégé est pass, warn ou fail. Les paramètres d'adaptateur invalides, inconnus ou mal orthographiés échouent avant les diagnostics ou le démarrage du runtime. Un descripteur résolu sans schéma de paramètres accepte uniquement une map de paramètres vide.

Consommer les résultats et gérer les erreurs

Chaque invocation qui atteint la limite d'adaptateur retourne un RunResult normalisé, même quand l'invocation du harness elle-même a échoué. Inspectez les champs d'échec avant de lire la sortie :

result = await fabric.run(config, base_dir=base, input="Review the changes.")

if result.status == "succeeded":
    use_output(result.output, result.artifacts, result.telemetry)
else:
    handle_failure(result.status, result.error, result.events)  # failed, cancelled, ...

Traitez status == "succeeded" comme le seul succès. Les autres valeurs terminales (failed, cancelled) sont infructueuses, donc branchez sur status, non sur error. Lisez status, error et events avant de traiter output.
Capturez les références artifacts et telemetry comme la preuve retournée pour les plateformes et les évaluations. Stockez et journalisez runtime_id, invocation_id et request_id séparément comme des chaînes opaques.
Capturez les sous-classes FabricError pour les défaillances de cycle de vie qui empêchent un résultat normalisé : FabricConfigError, FabricCapabilityError, FabricRuntimeError, FabricStateError et FabricNativeUnavailableError.
Le consommateur possède les tentatives et la politique d'échec ; NeMo Fabric ne réessaie pas par défaut. run(...) et les runtimes async with tentent un nettoyage automatique, préférez-les à stop() manuel — mais l'arrêt n'est pas garanti : stop(), y compris l'appel automatique quand un bloc async with se termine, peut lever FabricRuntimeError. Lors d'une sortie normale, cette erreur se propage ; après une erreur d'invocation, l'échec de nettoyage est attaché à l'exception d'origine. Soyez prêt à gérer un échec d'arrêt.

Consultez results-and-errors.md pour l'inventaire complet des champs de résultats et d'erreurs, et sdk-api-inventory.md pour savoir quand utiliser chaque méthode Fabric et Runtime.

Tester et valider l'intégration

Écrivez des tests d'intégration ciblés qui construisent la FabricConfig du consommateur, affirment que plan(...) sélectionne l'adaptateur et les capacités attendus, et — où un harness et des identifiants sont disponibles — exécutez une invocation et affirmez le RunResult status et la preuve.
plan(...) est sans identifiants — utilisez-le comme la porte CI qui valide la sélection d'adaptateur et le routage de capacités sans modèle ni secrets. doctor(...) s'exécute aussi sans appeler un modèle, mais il vérifie les exigences d'environnement déclarées (telles que les variables de clé API requises) et retourne fail quand elles ne sont pas définies, alors exécutez-le où l'environnement est provisionné et lisez ses résultats par vérification.
Exécutez les commandes de construction et de test propres du projet consommateur. Pour un checkout de source de NeMo Fabric, just build-all reconstruit l'extension native et just test-python exécute la suite Python.
Confirmez que la config typée est passée directement à NeMo Fabric et qu'aucune importation non-publique n'a été ajoutée.

Liste de contrôle

[ ] L'objet de configuration consommatrice est traduit directement en FabricConfig en mémoire.
[ ] Seuls les symboles publics nemo_fabric sont importés ; pas de _native ni d'internals d'adaptateur.
[ ] La config consommatrice est construite en mémoire et passée directement à NeMo Fabric.
[ ] Le bon cycle de vie est choisi : run(...) pour une invocation unique, start_runtime(...) avec async with pour multi-tour, invoke_openai_stream(...) pour les chunks OpenAI limitées par descripteur, ou invoke_stream(...) pour l'ATOF NeMo Relay brut.
[ ] plan(...) et doctor(...) valident la sélection d'adaptateur, les capacités et l'environnement avant exécution.
[ ] Installation, dépendances d'adaptateur et identifiants sont possédés par l'environnement, non le code consommateur.
[ ] Le RunResult status, error et events sont inspectés avant output ; artifacts et telemetry sont capturés.
[ ] Les sous-classes FabricError sont gérées, y compris un FabricRuntimeError levé par l'arrêt ; le nettoyage est délégué à run(...) ou async with (tenté, non garanti).
[ ] Les ID de corrélation sont stockés et journalisés comme des chaînes opaques.
[ ] Les tests d'intégration ciblés réussissent et la validation NeMo Fabric (plan/doctor, tests) réussit.

Documentation associée

Liez à ces sources canoniques au lieu de les dupliquer :

Guide SDK Python
Aperçu NeMo Fabric et guide d'installation
Référence API générée (index d'API publique ; les type stubs nemo_fabric installés font autorité pour les signatures exactes, champs et défauts) : client, runtime, streaming OpenAI natif, streaming Relay, modèles, types, erreurs
Exemple de config en mémoire canonique : examples/code_review_agent
Intégration de plateforme et de harness d'évaluation : examples/harbor et nemo_fabric.integrations.harbor. Harbor construit une config typée à partir d'entrées d'agent explicites et la transporte à l'intérieur d'une spécification de lancement transient privée à la limite du processus de tâche. Suivez l'exemple de révision de code pour le code d'intégration consommatrice ; la représentation de transport de Harbor est un contrat de limite de processus interne.

nemo-fabric-integrate