#!/usr/bin/env python3
"""
Generate OpenAPI schemas for GPT and full/dev access.
This script reads the combined OpenAPI spec and generates:
1. openapi-gpt.json - GPT-facing schema (≤30 operations)
2. openapi-full.json - Full schema for dev/MCP (unrestricted)
The GPT schema is generated by applying a denylist of dev/ops endpoints
and adding the consolidated entity endpoint.
"""
import json
import copy
from pathlib import Path
from typing import Any
# Configuration
GPT_MAX_OPERATIONS = 30
GPT_TARGET_OPERATIONS = 29 # Leave 1 spare slot
# Endpoints to REMOVE from GPT schema (dev/ops endpoints)
GPT_DENYLIST = {
("get", "/api/koi/stats"), # Dev/info endpoint
("get", "/api/koi/health"), # Dev/ops endpoint
}
# Future endpoints that should never appear in GPT schema
# (for validation purposes even if not in current schema)
GPT_FORBIDDEN_PATTERNS = [
"/ops/",
"/utils/",
"/debug/",
"/admin/",
"/internal/",
]
def load_openapi_spec(path: Path) -> dict[str, Any]:
"""Load OpenAPI spec from JSON file."""
with open(path, "r") as f:
return json.load(f)
def save_openapi_spec(spec: dict[str, Any], path: Path) -> None:
"""Save OpenAPI spec to JSON file with consistent formatting."""
with open(path, "w") as f:
json.dump(spec, f, indent=2)
print(f"Saved: {path}")
def count_operations(spec: dict[str, Any]) -> int:
"""Count the number of operations in an OpenAPI spec."""
count = 0
for path, methods in spec.get("paths", {}).items():
for method in methods:
if method.lower() in ["get", "post", "put", "patch", "delete"]:
count += 1
return count
def list_operations(spec: dict[str, Any]) -> list[tuple[str, str, str]]:
"""List all operations as (method, path, operationId) tuples."""
operations = []
for path, methods in spec.get("paths", {}).items():
for method, details in methods.items():
if method.lower() in ["get", "post", "put", "patch", "delete"]:
op_id = details.get("operationId", "unknown")
operations.append((method.upper(), path, op_id))
return sorted(operations, key=lambda x: (x[1], x[0]))
def filter_paths(spec: dict[str, Any], predicate) -> dict[str, Any]:
"""Return a shallow copy of spec with only paths matching predicate(path, methods)."""
out = dict(spec)
out["paths"] = {}
for path, methods in spec.get("paths", {}).items():
if predicate(path, methods):
out["paths"][path] = methods
return out
def set_single_server(spec: dict[str, Any], url: str, description: str) -> None:
"""Set a single server URL for an OpenAPI spec."""
spec["servers"] = [{"url": url, "description": description}]
def get_consolidated_entity_endpoint() -> dict[str, Any]:
"""
Return the consolidated POST /api/koi/entity endpoint definition.
This endpoint combines resolve, neighborhood, and documents queries
into a single operation via query_type parameter.
"""
return {
"post": {
"tags": ["Knowledge - Entity"],
"summary": "Entity relationship queries",
"description": (
"Entity graph query. Set query_type to one of: resolve | neighborhood | documents. "
"Provide label (to resolve) or uri (if already known)."
),
"operationId": "koi_entity_query",
"requestBody": {
"content": {
"application/json": {
"schema": {
"type": "object",
"required": ["query_type"],
"properties": {
"query_type": {
"type": "string",
"enum": ["resolve", "neighborhood", "documents"],
"description": "Type of entity query to perform"
},
"label": {
"type": "string",
"description": "Entity label to look up (e.g., 'ethereum', 'regen commons')"
},
"uri": {
"type": "string",
"description": "Entity URI (preferred if known from previous resolve)"
},
"type_hint": {
"type": "string",
"description": "Optional type hint for disambiguation (e.g., 'TECHNOLOGY', 'ORGANIZATION')"
},
"direction": {
"type": "string",
"enum": ["out", "in", "both"],
"default": "both",
"description": "Edge direction for neighborhood query"
},
"limit": {
"type": "integer",
"minimum": 1,
"maximum": 100,
"default": 20,
"description": "Maximum results to return"
}
}
}
}
},
"required": True
},
"responses": {
"200": {
"description": "Entity query results",
"content": {
"application/json": {
"schema": {
"type": "object",
"properties": {
"data": {
"type": "object",
"description": "Entity query payload (shape depends on query_type)",
"properties": {
"query_type": {"type": "string"},
"query_label": {"type": "string"},
"winner": {"type": "object"},
"alternatives": {"type": "array", "items": {"type": "object"}},
"is_polysemy": {"type": "boolean"}
}
},
"request_id": {"type": "string"},
"data_source": {"type": "string"},
"citations": {"type": "array", "items": {"type": "object"}},
"warnings": {"type": "array", "items": {"type": "string"}},
"errors": {"type": "array", "items": {"type": "object"}},
"as_of": {"type": "object"},
"tool_trace": {"type": "array", "items": {"type": "object"}}
},
"required": ["data", "request_id"]
},
"example": {
"data": {
"query_type": "resolve",
"query_label": "Ethereum",
"variant_count": 3,
"winner": {
"uri": "https://regen.network/tech/b35599829c6dc565",
"entity_text": "Ethereum",
"entity_type": "TECHNOLOGY"
},
"alternatives": [
{"uri": "https://regen.network/project/a3b7abd349ff09a5", "entity_text": "Ethereum", "entity_type": "PROJECT"}
],
"is_polysemy": True
},
"request_id": "550e8400-e29b-41d4-a716-446655440999",
"data_source": "koi-derived",
"citations": [],
"warnings": [],
"errors": [],
"as_of": {"koi": {"corpus_version": "2026-01-01", "indexed_at": "2026-01-01T00:00:00Z"}},
"tool_trace": [{"tool": "entity_resolve", "params_summary": "label=Ethereum,limit=5"}]
}
}
}
}
}
}
}
def ensure_koi_health_trailing_slash(spec: dict[str, Any]) -> None:
"""
KOI health responds at /api/koi/health/ and redirects from /api/koi/health.
Some GPT Action callers do not reliably follow redirects, so model /api/koi/health/.
"""
old_path = "/api/koi/health"
new_path = "/api/koi/health/"
if old_path in spec.get("paths", {}) and new_path not in spec["paths"]:
spec["paths"][new_path] = spec["paths"].pop(old_path)
def generate_gpt_schema(combined_spec: dict[str, Any]) -> dict[str, Any]:
"""
Generate GPT-facing OpenAPI schema with ≤30 operations.
- Removes dev/ops endpoints from denylist
- Adds consolidated entity endpoint
- Updates info to reflect GPT-facing nature
"""
gpt_spec = copy.deepcopy(combined_spec)
# Update info
gpt_spec["info"]["title"] = "Regen Network API (GPT)"
gpt_spec["info"]["description"] = (
"GPT-optimized API for Regen Network. "
"Combines blockchain ledger data and knowledge base search.\n\n"
"**IMPORTANT**: Call /regen-api/summary first to understand available endpoints.\n\n"
+ combined_spec["info"].get("description", "")
)
# Remove denylisted endpoints
paths_to_remove = []
for path, methods in gpt_spec.get("paths", {}).items():
methods_to_remove = []
for method in methods:
if method.lower() in ["get", "post", "put", "patch", "delete"]:
key = (method.lower(), path)
if key in GPT_DENYLIST:
methods_to_remove.append(method)
for method in methods_to_remove:
del gpt_spec["paths"][path][method]
print(f" Removed: {method.upper()} {path}")
# If path has no methods left, mark for removal
if not any(m.lower() in ["get", "post", "put", "patch", "delete"]
for m in gpt_spec["paths"][path]):
paths_to_remove.append(path)
for path in paths_to_remove:
del gpt_spec["paths"][path]
# Add consolidated entity endpoint
gpt_spec["paths"]["/api/koi/entity"] = get_consolidated_entity_endpoint()
print(f" Added: POST /api/koi/entity (consolidated)")
return gpt_spec
def generate_full_schema(combined_spec: dict[str, Any]) -> dict[str, Any]:
"""
Generate full OpenAPI schema for dev/MCP (unrestricted).
- Keeps all endpoints including dev/ops
- Adds consolidated entity endpoint for consistency
- Updates info to reflect dev/MCP nature
"""
full_spec = copy.deepcopy(combined_spec)
# Update info
full_spec["info"]["title"] = "Regen Network API (Full)"
full_spec["info"]["description"] = (
"Full API for Regen Network (dev/MCP). "
"Includes all endpoints including dev/ops tools.\n\n"
+ combined_spec["info"].get("description", "")
)
# Add consolidated entity endpoint (for consistency with GPT schema)
full_spec["paths"]["/api/koi/entity"] = get_consolidated_entity_endpoint()
print(f" Added: POST /api/koi/entity (consolidated)")
return full_spec
def generate_split_action_schemas(
combined_spec: dict[str, Any],
gpt_spec: dict[str, Any],
full_spec: dict[str, Any],
) -> tuple[dict[str, Any], dict[str, Any]]:
"""
Generate two Action-friendly schemas (each <=30 ops) for a two-Action GPT setup:
- Ledger Action: /regen-api/* on https://regen.gaiaai.xyz
- KOI Action: /api/koi/* on https://registry.regen.gaiaai.xyz
"""
ledger_spec = filter_paths(gpt_spec, lambda path, methods: path.startswith("/regen-api/"))
set_single_server(ledger_spec, "https://regen.gaiaai.xyz", "Production API (Ledger Action)")
# KOI Action: keep the GPT-safe KOI surface (no stats/health/debug).
koi_spec = filter_paths(gpt_spec, lambda path, methods: path.startswith("/api/koi/"))
set_single_server(koi_spec, "https://registry.regen.gaiaai.xyz", "Production API (KOI Action)")
return ledger_spec, koi_spec
def print_operation_summary(spec: dict[str, Any], name: str) -> None:
"""Print a summary of operations in the spec."""
ops = list_operations(spec)
count = len(ops)
print(f"\n{name}:")
print(f" Total operations: {count}")
if count > GPT_MAX_OPERATIONS and "GPT" in name:
print(f" ⚠️ EXCEEDS GPT LIMIT OF {GPT_MAX_OPERATIONS}!")
print("\n Operations:")
for method, path, op_id in ops:
print(f" {method:6} {path:50} ({op_id})")
def main() -> None:
"""Main entry point."""
script_dir = Path(__file__).parent
repo_root = script_dir.parent
combined_path = repo_root / "openapi-combined.json"
gpt_path = repo_root / "openapi-gpt.json"
full_path = repo_root / "openapi-full.json"
ledger_action_path = repo_root / "openapi-gpt-ledger.json"
koi_action_path = repo_root / "openapi-gpt-koi.json"
print("=" * 60)
print("OpenAPI Schema Generator")
print("=" * 60)
# Load combined spec
print(f"\nLoading: {combined_path}")
combined_spec = load_openapi_spec(combined_path)
print_operation_summary(combined_spec, "Source (openapi-combined.json)")
# Generate GPT schema
print("\n" + "-" * 60)
print("Generating GPT schema...")
gpt_spec = generate_gpt_schema(combined_spec)
save_openapi_spec(gpt_spec, gpt_path)
print_operation_summary(gpt_spec, "GPT Schema (openapi-gpt.json)")
# Generate full schema
print("\n" + "-" * 60)
print("Generating full schema...")
full_spec = generate_full_schema(combined_spec)
save_openapi_spec(full_spec, full_path)
print_operation_summary(full_spec, "Full Schema (openapi-full.json)")
# Generate split Action schemas (two Action domains)
print("\n" + "-" * 60)
print("Generating split Action schemas (ledger-only + koi-only)...")
ledger_spec, koi_spec = generate_split_action_schemas(combined_spec, gpt_spec, full_spec)
save_openapi_spec(ledger_spec, ledger_action_path)
save_openapi_spec(koi_spec, koi_action_path)
print_operation_summary(ledger_spec, "Ledger Action Schema (openapi-gpt-ledger.json)")
print_operation_summary(koi_spec, "KOI Action Schema (openapi-gpt-koi.json)")
# Summary
gpt_count = count_operations(gpt_spec)
full_count = count_operations(full_spec)
print("\n" + "=" * 60)
print("Summary")
print("=" * 60)
print(f" GPT schema: {gpt_count} operations (limit: {GPT_MAX_OPERATIONS}, target: {GPT_TARGET_OPERATIONS})")
print(f" Full schema: {full_count} operations (unrestricted)")
if gpt_count <= GPT_MAX_OPERATIONS:
print(f"\n✅ GPT schema is within the {GPT_MAX_OPERATIONS}-operation limit")
else:
print(f"\n❌ GPT schema EXCEEDS the {GPT_MAX_OPERATIONS}-operation limit!")
exit(1)
if __name__ == "__main__":
main()