Grabadora Asset Core Puerta de Decisión
Docs
Documentos de Decision Gate

Evaluación de puertas determinista, reproducible con decisiones auditables.

Documentación de Asset Core

Manual de OpenAPI Tipado

A primera vista

Qué: Importar OpenAPI en artefactos de proveedor tipados y gestionar versiones del ciclo de vida. Por qué: Obtener artefactos de capacidad/ejecución estrictos sin construir un servidor MCP personalizado. Quién: Operadores que integran APIs respaldadas por OpenAPI en Decision Gate. Requisitos previos: provider_schema_authoring.md, condition_authoring.md, openapi_reference_library.md

Flujo de Herramienta

Herramientas de ciclo de vida tipadas:

  1. typed_providers_import
  2. typed_providers_register
  3. typed_providers_list
  4. typed_providers_get
  5. typed_providers_activate
  6. typed_providers_deprecate

typed_providers_import es la ruta OSS principal para flujos de trabajo impulsados por OpenAPI. Compila artefactos de contrato/runtime deterministas y registra una versión del ciclo de vida en una sola operación.

Todas las herramientas de ciclo de vida tipadas requieren campos de alcance explícitos:

  • id_de_inquilino
  • namespace_id

Importar Perillas y Modos de Conformidad

typed_providers_import admite perillas limitadas para importaciones deterministas:

  • credential_bindings: map of OpenAPI security scheme id -> structured binding (locator + value_render requerido; display_name opcional)
  • openapi_semantics_mode: auto | oas30 | oas31
  • media_support_mode: json_only | all_media
  • external_ref_mode: solo_archivo_local | lista_de_permitidos_de_red
  • openapi_conformance_mode: estricto | auditoría

strict rechaza construcciones OpenAPI no soportadas. audit permite el registro mientras emite hallazgos no soportados determinísticos en conformance_summary. credential_bindings es siempre requerido por la carga útil de la herramienta; los campos omitidos fallan en la decodificación/validación. Proporcione un mapa vacío para importaciones no autenticadas y mapeos explícitos para operaciones aseguradas. Los esquemas de localización permitidos son secret://... y env://... solamente; la resolución de env://... está deshabilitada por defecto y requiere dev.allow_dev_env_credentials=true. value_render es explícito y falla en cerrado:

  • {"mode":"identity"} utiliza el secreto en bruto tal como está.
  • {"mode":"prefix","prefix":"Token "} prepends a deterministic prefix. Notas secretas de aprovisionamiento:
  • Store raw provider credentials in the encrypted secret store and reference by secret://... localizador.
  • If keyring is unavailable/headless, set DECISION_GATE_SECRETS_PASSPHRASE before running secrets commands so the store can be unlocked. JSON object and array-complex response schemas must declare x-decision-gate.projections; missing projection metadata fails import. Legacy response_projection_mode input is removed and rejected. Projection metadata is evaluated on normalized/resolved schemas, so component-level metadata via $ref is first-class. No duplique los esquemas de respuesta en línea únicamente para llevar proyecciones.

El comportamiento semántico se aplica en el momento de la importación:

  • auto: inferir semántica del encabezado openapi (3.0.x o 3.1.x).
  • oas30: el encabezado debe ser 3.0.x; los arreglos de tipo JSON Schema son rechazados.
  • oas31: el encabezado debe ser 3.1.x; la palabra clave heredada nullable es rechazada.

Perfil de Ejecución y Semántica de Deriva

Los registros de ciclo de vida tipados y los perfiles de tiempo de ejecución llevan metadatos de resumen:

  • source_digest: resumen de la entrada de origen importada
  • profile_digest: resumen del perfil de ejecución generado
  • contract_hash: hash canónico del contrato del proveedor generado

typed_providers_get puede calcular la deriva con observed_source_digest + observed_profile_digest y devuelve drift_status cuando se detectan discrepancias.

Configuración del Proveedor Tipado

Los proveedores tipados se configuran con artefactos preconstruidos:

[[providers]]
name = "typed_asset_api"
type = "typed"
capabilities_path = "contracts/typed_asset_api.json"
runtime_profile_path = "profiles/typed_asset_api_runtime.json"
typed_protocol = "openapi_http"

Ejemplo de validación:

[server]
transport = "http"
bind = "127.0.0.1:4000"
mode = "strict"

[server.auth]
mode = "local_only"

[namespace]
allow_default = true
default_tenants = [1]

[trust]
default_policy = "audit"
min_lane = "verified"

[evidence]
allow_raw_values = false
require_provider_opt_in = true

[schema_registry]
type = "sqlite"
path = "decision-gate-registry.db"

[schema_registry.acl]
allow_local_only = true
require_signing = false

[run_state_store]
type = "sqlite"
path = "decision-gate.db"
journal_mode = "wal"
sync_mode = "full"
busy_timeout_ms = 5000

[[providers]]
name = "time"
type = "builtin"

[[providers]]
name = "env"
type = "builtin"

[[providers]]
name = "json"
type = "builtin"
config = { root = "./evidence", root_id = "evidence-root", max_bytes = 1048576, allow_yaml = true }

[[providers]]
name = "http"
type = "builtin"

[[providers]]
name = "rest"
type = "builtin"

[[providers]]
name = "typed_asset_api"
type = "typed"
capabilities_path = "contracts/typed_asset_api.json"
runtime_profile_path = "profiles/typed_asset_api_runtime.json"
typed_protocol = "openapi_http"

Carga Útil de Importación de OpenAPI

Forma mínima de carga útil de importación:

{
  "provider_id": "typed_asset_api",
  "version": "v1",
  "openapi": {
    "openapi": "3.1.0",
    "paths": {
      "/assets": {
        "get": {
          "operationId": "listAssets",
          "responses": {
            "200": {
              "description": "ok",
              "content": {
                "application/json": {
                  "schema": {
                    "type": "object",
                    "properties": {
                      "next_cursor": { "type": "string" }
                    },
                    "required": ["next_cursor"],
                    "additionalProperties": false,
                    "x-decision-gate": {
                      "projections": [
                        {
                          "id": "next_cursor",
                          "pointer": "/next_cursor",
                          "schema": { "type": "string" }
                        }
                      ]
                    }
                  }
                }
              }
            }
          }
        }
      }
    }
  },
  "operation_allowlist": ["listAssets"],
  "credential_bindings": {},
  "allow_unsafe_methods": false,
  "timeout_ms": 5000,
  "max_response_bytes": 1048576,
  "activate": true
}

Verificación del Ciclo de Vida Ejecutable

Este bloque importa un documento OpenAPI mínimo, valida la visibilidad de lista/obtención y afirma los campos de metadatos de contrato/tiempo de ejecución.

import json
import os
from urllib import request


def call_tool(endpoint: str, tool_name: str, arguments: dict) -> dict:
    payload = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "tools/call",
        "params": {"name": tool_name, "arguments": arguments},
    }
    req = request.Request(
        endpoint,
        data=json.dumps(payload).encode("utf-8"),
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    with request.urlopen(req, timeout=20) as resp:
        body = json.loads(resp.read().decode("utf-8"))
    if "error" in body:
        raise RuntimeError(f"tool call failed: {body['error']}")
    content = body.get("result", {}).get("content", [])
    if not content or "json" not in content[0]:
        raise RuntimeError(f"unexpected tool result envelope: {body}")
    return content[0]["json"]


endpoint = os.environ.get("DG_ENDPOINT", "http://127.0.0.1:8080/rpc")
provider_id = "docs_typed_asset_api"
version = "v1"

openapi_doc = {
    "openapi": "3.1.0",
    "components": {
        "schemas": {
            "AssetListResponse": {
                "type": "object",
                "properties": {"next_cursor": {"type": "string"}},
                "required": ["next_cursor"],
                "additionalProperties": False,
                "x-decision-gate": {
                    "projections": [
                        {
                            "id": "next_cursor",
                            "pointer": "/next_cursor",
                            "schema": {"type": "string"},
                        }
                    ]
                },
            }
        }
    },
    "paths": {
        "/assets": {
            "get": {
                "operationId": "listAssets",
                "responses": {
                    "200": {
                        "description": "ok",
                        "content": {
                            "application/json": {
                                "schema": {
                                    "$ref": "#/components/schemas/AssetListResponse"
                                }
                            }
                        },
                    }
                },
            }
        }
    },
}

import_response = call_tool(
    endpoint,
    "typed_providers_import",
    {
        "tenant_id": 1,
        "namespace_id": 1,
        "provider_id": provider_id,
        "version": version,
        "openapi": openapi_doc,
        "operation_allowlist": ["listAssets"],
        "credential_bindings": {},
        "allow_unsafe_methods": False,
        "timeout_ms": 5000,
        "max_response_bytes": 1048576,
        "openapi_conformance_mode": "strict",
        "activate": True,
    },
)
assert import_response["provider_id"] == provider_id, import_response
assert import_response["version"] == version, import_response
assert import_response["register_outcome"] in {"registered", "updated"}, import_response
assert import_response["active_version"] == version, import_response
assert import_response["source_digest"]["algorithm"] == "sha256", import_response
assert isinstance(import_response["source_digest"]["value"], str), import_response
assert import_response["source_digest"]["value"], import_response
assert import_response["profile_digest"]["algorithm"] == "sha256", import_response
assert isinstance(import_response["profile_digest"]["value"], str), import_response
assert import_response["profile_digest"]["value"], import_response
assert import_response["contract_hash"]["algorithm"] == "sha256", import_response
assert isinstance(import_response["contract_hash"]["value"], str), import_response
assert import_response["contract_hash"]["value"], import_response
assert import_response["operation_count"] >= 1, import_response

list_response = call_tool(endpoint, "typed_providers_list", {"tenant_id": 1, "namespace_id": 1})
items = list_response.get("items", [])
assert any(item.get("provider_id") == provider_id for item in items), list_response

get_response = call_tool(
    endpoint,
    "typed_providers_get",
    {"tenant_id": 1, "namespace_id": 1, "provider_id": provider_id, "version": version},
)
assert get_response["provider_id"] == provider_id, get_response
assert get_response["selected_version"] == version, get_response
assert get_response["contract"]["transport"] == "typed", get_response
assert get_response["runtime_profile"]["provider_id"] == provider_id, get_response
assert get_response["record"]["source_digest"] == import_response["source_digest"], get_response
assert get_response["record"]["profile_digest"] == import_response["profile_digest"], get_response
assert get_response["runtime_profile"]["source_digest"] == import_response["source_digest"], get_response
assert get_response["runtime_profile"]["profile_digest"] == import_response["profile_digest"], get_response
assert get_response["drift_status"] is None, get_response
operations = get_response["runtime_profile"].get("operations", [])
assert isinstance(operations, list) and len(operations) > 0, get_response
assert isinstance(operations[0].get("check_id"), str), get_response

Notas

  • Mantener operation_allowlist explícito y estable para preservar los IDs de verificación determinísticos.
  • Utiliza los campos de resumen typed_providers_get para el seguimiento de desviaciones en los flujos de trabajo de implementación.
  • Utiliza typed_providers_deprecate con semántica de retroceso para cambios de ciclo de vida controlados.