OpenAPI adapter (Python)

May 1, 2026 · View on GitHub

The asap.adapters.openapi package derives ASAP skills, JSON Schemas, and a task.request handler from an OpenAPI 3.0 or 3.1 document, then proxies each invocation to the upstream HTTP API described by the spec.

Requires the optional extra:

uv add 'asap-protocol[openapi]'
# or
pip install 'asap-protocol[openapi]'

This pulls in openapi-pydantic for validation and parsing.

Architecture

OpenAPI document (URL or local .json)
        │
        ▼
  load_spec()  ──► openapi-pydantic root model (3.0 / 3.1)
        │
        ▼
  map_openapi_to_capabilities()
        │  operationId → skill id; parameters + JSON body → input schema;
        │  200/201 application/json → output schema; OA-010 execution kind
        ▼
  OpenAPIUpstreamHandler  ◄── httpx.AsyncClient (spec fetch + upstream calls)
        │
        ▼
  create_openapi_task_handler()  ──► HandlerRegistry["task.request"]
        │
        ▼
  OpenAPIAdapterBundle.manifest + .registry  ──► create_app()
  • One async entrypoint: create_from_openapi builds an OpenAPIAdapterBundle with a ready-to-use Manifest, HandlerRegistry, and metadata (require_webauthn_for, upstream_base_url).
  • Upstream base URL: Taken from servers[0].url when it is absolute (https://…). If the spec uses a relative server URL (e.g. /api/v3), the adapter resolves it with urllib.parse.urljoin against the document URL used to load the spec. Loading from a local file only with a relative server requires upstream_base_url=.
  • Execution modes (OA-010): The mapper classifies each operation as sync, streaming (response advertises text/event-stream), or async_polling (202 + Location). The manifest’s Capability.streaming flag is set if any operation is streaming; the current task.request path still performs a single HTTP round-trip and returns JSON (or wrapped non-object JSON). Treat streaming/async polling as metadata for future wiring, not full ASAP streaming handlers yet.
  • Identity & capability grants: create_app still provisions identity and capability HTTP routes. Use FreshSessionConfig and bundle.require_webauthn_for (from approval_strength) for approval/WebAuthn policy; registering CapabilityDefinition rows in app.state.capability_registry is separate from OpenAPI mapping.

Quick usage

Always pass a shared httpx.AsyncClient (timeout, proxies, mock transports, MTLS as you configure it):

import asyncio
import httpx

from asap.adapters.openapi import create_from_openapi
from asap.transport.server import create_app


async def main() -> None:
    async with httpx.AsyncClient(timeout=60.0, follow_redirects=True) as http:
        bundle = await create_from_openapi(
            spec_url="https://api.example.com/openapi.json",
            http_client=http,
            default_capabilities=["GET", "HEAD"],
            upstream_base_url="https://api.example.com",  # optional override
            manifest_id="urn:asap:agent:my-openapi-bridge",
            asap_endpoint="https://my-host.example/asap",
        )

    app = create_app(bundle.manifest, bundle.registry)


if __name__ == "__main__":
    asyncio.run(main())

Approval strength (OA-008) and registration

Map HTTP methods or operationId values to session vs webauthn, then feed the derived list into FreshSessionConfig:

from asap.auth.self_auth import FreshSessionConfig
from asap.transport.server import create_app

bundle = await create_from_openapi(
    spec_url="https://api.example.com/openapi.json",
    http_client=http,
    approval_strength={"GET": "session", "POST": "webauthn", "DELETE": "webauthn"},
)

app = create_app(
    bundle.manifest,
    bundle.registry,
    identity_fresh_session_config=FreshSessionConfig(
        require_webauthn_for=bundle.require_webauthn_for,
    ),
)

Details: Self-authorization prevention.

Upstream auth (OA-009)

Pass a callable (session) -> dict[str, str] to inject headers (for example Authorization); operation-defined header parameters still merge and may override callback keys:

def resolve_headers(session: object | None) -> dict[str, str]:
    _ = session
    return {"Authorization": "Bearer <token>"}


bundle = await create_from_openapi(
    spec_url="https://api.example.com/openapi.json",
    http_client=http,
    resolve_headers=resolve_headers,
)

Configuration reference

ParameterDescription
spec_urlHTTP(S) URL of the OpenAPI document (exactly one of spec_url or spec_path).
spec_pathLocal path to a UTF-8 JSON document (.json). YAML is not supported yet.
http_clienthttpx.AsyncClient used to fetch the spec (if URL) and for all upstream API calls.
upstream_base_urlOverrides inferred base URL from servers.
default_capabilities"all", a single method string ("GET"), a sequence of methods, or a callable (OpenAPIOperationContext) -> bool. See map_openapi_to_capabilities.
approval_strengthOptional dict[str, str] mapping HTTP verbs and/or operationId to session or webauthn; populates bundle.require_webauthn_for for WebAuthn-gated capability names.
resolve_headersOptional callback for extra request headers to the upstream.
manifest_idASAP manifest id (default urn:asap:agent:openapi-adapter).
manifest_nameOverrides manifest name (default: info.title).
asap_endpointValue stored on Manifest.endpoints.asap (advertised to clients).

Bundle fields

FieldMeaning
manifestManifest with skills mirroring the selected operations.
registryHandlerRegistry with task.request → OpenAPI proxy handler.
capabilitiesList of OpenAPICapability (HTTP method, path template, operation_id, execution kind, etc.).
require_webauthn_forCapability names requiring WebAuthn when approval_strength is set.
upstream_base_urlResolved base URL used by the handler.

Runnable example and compliance

The repo includes examples/openapi_petstore/: bundled PetStore-shaped fragment, Compliance Harness v2 in-process, and a call to findPetsByStatus. By default the upstream is mocked for reliability; use --live for the public PetStore URL (network; remote service may error).

uv sync --extra openapi
uv run python examples/openapi_petstore/main.py

See also Compliance testing and CI compliance gate.

Common pitfalls

Authentication and secrets

  • The adapter does not read OpenAPI securitySchemes automatically. Use resolve_headers (Bearer, API keys, custom headers) or a dedicated gateway in front of the upstream.
  • Never embed tokens in source; load from environment or a secret store and build headers at runtime.

Relative servers and local specs

If the spec only declares servers: [{ url: "/api/v3" }] and you load from disk, the adapter cannot infer an absolute origin. Pass upstream_base_url explicitly.

Polymorphism and complex schemas

  • Input/output mapping focuses on application/json and inlines internal #/components/schemas/ refs where possible. oneOf / anyOf at the top level of request or response bodies may not round-trip cleanly into strict JSON Schema validation everywhere; review generated Skill schemas before exposing them to untrusted callers.
  • Non-JSON success bodies are returned as a small wrapper dict (e.g. _text, _json) depending on Content-Type.

Large specifications

  • Every operation can become a skill; large APIs (hundreds of paths) produce large manifests and handler indexes. Use default_capabilities (e.g. only GET) or a callable filter to trim surface area. Consider a curated spec or a gateway that exposes a subset.

Public demo APIs

Third-party OpenAPI hosts may return 4xx/5xx or change without notice. Prefer contract tests with mocked httpx transports for CI; use live URLs for manual exploration only.

Streaming (SSE) vs ASAP streaming

Upstream text/event-stream is detected for metadata, but the default task.request handler does not bridge SSE into ASAP TaskStream envelopes. Document behavior if you expose streaming operations; a dedicated streaming handler may be required for full parity.