Typed OpenAPI Playbook

At a Glance

What: Import OpenAPI into typed provider artifacts and manage lifecycle versions. Why: Get strict capability/runtime artifacts without building a custom MCP server. Who: Operators integrating OpenAPI-backed APIs into Decision Gate. Prerequisites: provider_schema_authoring.md, condition_authoring.md, openapi_reference_library.md

Tool Flow

Typed lifecycle tools:

typed_providers_import
typed_providers_register
typed_providers_list
typed_providers_get
typed_providers_activate
typed_providers_deprecate

typed_providers_import is the primary OSS path for OpenAPI-driven workflows. It compiles deterministic contract/runtime artifacts and registers a lifecycle version in one operation.

All typed lifecycle tools require explicit scope fields:

tenant_id
namespace_id

Import Knobs and Conformance Modes

typed_providers_import supports bounded knobs for deterministic import:

credential_bindings: map of OpenAPI security scheme id -> structured binding (locator + value_render required; display_name optional)
openapi_semantics_mode: auto | oas30 | oas31
media_support_mode: json_only | all_media
external_ref_mode: local_file_only | network_allowlist
openapi_conformance_mode: strict | audit

strict rejects unsupported OpenAPI constructs. audit permits registration while emitting deterministic unsupported findings in conformance_summary. credential_bindings is always required by the tool payload; omitted fields fail decode/validation. Provide an empty map for unauthenticated imports and explicit mappings for secured operations. Allowed locator schemes are secret://... and env://... only; env://... resolution is disabled by default and requires dev.allow_dev_env_credentials=true. value_render is explicit and fail-closed:

{"mode":"identity"} uses the raw secret as-is.
{"mode":"prefix","prefix":"Token "} prepends a deterministic prefix. Secret provisioning notes:
Store raw provider credentials in the encrypted secret store and reference by secret://... locator.
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. Do not duplicate inline response schemas purely to carry projections.

Semantics behavior is enforced at import time:

auto: infer semantics from the openapi header (3.0.x or 3.1.x).
oas30: header must be 3.0.x; JSON Schema type arrays are rejected.
oas31: header must be 3.1.x; legacy nullable keyword is rejected.

Runtime Profile and Drift Semantics

Typed lifecycle records and runtime profiles carry digest metadata:

source_digest: digest of imported source input
profile_digest: digest of generated runtime profile
contract_hash: canonical hash of generated provider contract

typed_providers_get can compute drift with observed_source_digest + observed_profile_digest and returns drift_status when mismatches are detected.

Typed Provider Config

Typed providers are configured with prebuilt artifacts:

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

Validation example:

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

OpenAPI Import Payload

Minimal import payload shape:

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

Runnable Lifecycle Check

This block imports a minimal OpenAPI document, validates list/get visibility, and asserts contract/runtime metadata fields.

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

Notes

Keep operation_allowlist explicit and stable to preserve deterministic check IDs.
Use typed_providers_get digest fields for drift tracking in deployment workflows.
Use typed_providers_deprecate with rollback semantics for controlled lifecycle changes.