دليل OpenAPI المكتوب

نظرة سريعة

ما: استيراد OpenAPI إلى قطع مزود مكتوبة وإدارة إصدارات دورة الحياة. لماذا: الحصول على قطع قدرة/تشغيل صارمة دون بناء خادم MCP مخصص. من: المشغلون الذين يدمجون واجهات برمجة التطبيقات المدعومة من OpenAPI في Decision Gate. المتطلبات المسبقة: provider_schema_authoring.md, condition_authoring.md, openapi_reference_library.md

تدفق الأداة

أدوات دورة الحياة المكتوبة:

typed_providers_import
typed_providers_register
typed_providers_list
typed_providers_get
typed_providers_activate
typed_providers_deprecate

typed_providers_import هو المسار الرئيسي مفتوح المصدر لعمليات OpenAPI المدفوعة. يقوم بتجميع العناصر التعاقدية / التشغيلية الحتمية ويسجل إصدار دورة الحياة في عملية واحدة.

تتطلب جميع أدوات دورة الحياة المكتوبة حقول نطاق صريحة:

معرف_المستأجر
namespace_id

استيراد الأزرار وأنماط التوافق

typed_providers_import يدعم الأزرار المحدودة للاستيراد الحتمي:

credential_bindings: map of OpenAPI security scheme id -> structured binding (locator + value_render مطلوب؛ display_name اختياري)
openapi_semantics_mode: تلقائي | oas30 | oas31
media_support_mode: json_only | all_media
external_ref_mode: local_file_only | network_allowlist
openapi_conformance_mode: صارم | تدقيق

strict يرفض البنى غير المدعومة في OpenAPI. audit يسمح بالتسجيل بينما يصدر نتائج غير مدعومة حتمية في conformance_summary. credential_bindings مطلوب دائمًا من حمولة الأداة؛ الحقول المفقودة تفشل في فك التشفير/التحقق. قدم خريطة فارغة للاستيرادات غير المصرح بها وتعيينات صريحة للعمليات المؤمنة. المخططات المسموح بها هي secret://... و env://... فقط؛ يتم تعطيل حل env://... بشكل افتراضي ويتطلب dev.allow_dev_env_credentials=true. value_render صريح ويفشل في الإغلاق:

{"mode":"identity"} يستخدم السر الخام كما هو.
{"mode":"prefix","prefix":"Token "} prepends a deterministic prefix. ملاحظات التزويد السرية:
Store raw provider credentials in the encrypted secret store and reference by secret://... محدد.
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. لا تقم بتكرار مخططات الاستجابة المضمنة فقط لنقل التوقعات.

سلوك الدلالات يُفرض في وقت الاستيراد:

auto: استنتاج الدلالات من رأس openapi (3.0.x أو 3.1.x).
oas30: يجب أن يكون العنوان 3.0.x؛ يتم رفض مصفوفات نوع JSON Schema.
oas31: يجب أن يكون العنوان 3.1.x؛ الكلمة الرئيسية القديمة nullable مرفوضة.

ملف وقت التشغيل ودلالات الانحراف

تحتوي سجلات دورة الحياة المكتوبة وملفات تعريف وقت التشغيل على بيانات وصفية ملخصة:

source_digest: ملخص الإدخال المصدر المستورد
profile_digest: ملخص لملف التشغيل الذي تم إنشاؤه
contract_hash: تجزئة معيارية لعقد المزود الذي تم إنشاؤه

يمكن لـ typed_providers_get حساب الانحراف باستخدام observed_source_digest + observed_profile_digest ويعيد drift_status عند اكتشاف عدم تطابق.

تكوين مزود الكتابة

تم تكوين موفري الكتابة باستخدام قطع أثرية مسبقة البناء:

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

مثال على التحقق:

[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

شكل الحمولة الدنيا للاستيراد:

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

فحص دورة حياة القابل للتشغيل

هذا الكتلة تستورد وثيقة OpenAPI بسيطة، تتحقق من رؤية القائمة/الحصول، وتؤكد على حقول بيانات العقد/وقت التشغيل.

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

ملاحظات

حافظ على operation_allowlist واضحًا ومستقرًا للحفاظ على معرفات التحقق الحتمية.
استخدم حقول typed_providers_get للتتبع في سير العمل للنشر.
استخدم typed_providers_deprecate مع دلالات التراجع لتغييرات دورة الحياة المتحكم بها.