yunxiafei/openclaw-mission-control
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(api): enhance OpenAPI documentation with additional endpoints and examples

Browse Source
This commit is contained in:
Abhimanyu Saharan
2026-02-15 02:57:06 +05:30
parent ae17facf88
commit 07df7d8962
2 changed files with 338 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
backend/app/api/task_custom_fields.py
View File

@@ -32,7 +32,7 @@ if TYPE_CHECKING:
from sqlmodel.ext.asyncio.session import AsyncSession
router = APIRouter(prefix="/organizations/me/custom-fields", tags=["org-custom-fields"])
router = APIRouter(prefix="/organizations/me/custom-fields", tags=["custom-fields"])
SESSION_DEP = Depends(get_session)
ORG_MEMBER_DEP = Depends(require_org_member)
ORG_ADMIN_DEP = Depends(require_org_admin)

338
backend/app/main.py
View File

@@ -3,9 +3,10 @@
from __future__ import annotations
from contextlib import asynccontextmanager
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Any
from fastapi import APIRouter, FastAPI, status
from fastapi.openapi.utils import get_openapi
from fastapi.middleware.cors import CORSMiddleware
from fastapi_pagination import add_pagination
@@ -54,6 +55,78 @@ OPENAPI_TAGS = [
"Service liveness/readiness probes used by infrastructure and runtime checks."
),
},
{
"name": "agents",
"description": "Organization-level agent directory, lifecycle, and management operations.",
},
{
"name": "activity",
"description": "Activity feed and audit timeline endpoints across boards and operations.",
},
{
"name": "gateways",
"description": "Gateway management, synchronization, and runtime control operations.",
},
{
"name": "metrics",
"description": "Aggregated operational and board analytics metrics endpoints.",
},
{
"name": "organizations",
"description": "Organization profile, membership, and governance management endpoints.",
},
{
"name": "souls-directory",
"description": "Directory and lookup endpoints for agent soul templates and variants.",
},
{
"name": "skills",
"description": "Skills marketplace, install, uninstall, and synchronization endpoints.",
},
{
"name": "board-groups",
"description": "Board group CRUD, assignment, and grouping workflow endpoints.",
},
{
"name": "board-group-memory",
"description": "Shared memory endpoints scoped to board groups and grouped boards.",
},
{
"name": "boards",
"description": "Board lifecycle, configuration, and board-level management endpoints.",
},
{
"name": "board-memory",
"description": "Board-scoped memory read/write endpoints for persistent context.",
},
{
"name": "board-webhooks",
"description": "Board webhook registration, delivery config, and lifecycle endpoints.",
},
{
"name": "board-onboarding",
"description": "Board onboarding state, setup actions, and onboarding workflow endpoints.",
},
{
"name": "approvals",
"description": "Approval request, review, and status-tracking operations for board tasks.",
},
{
"name": "tasks",
"description": "Task CRUD, dependency management, and task workflow operations.",
},
{
"name": "custom-fields",
"description": "Organization custom-field definitions and board assignment endpoints.",
},
{
"name": "tags",
"description": "Tag catalog and task-tag association management endpoints.",
},
{
"name": "users",
"description": "User profile read/update operations and user-centric settings endpoints.",
},
{
"name": "agent",
"description": (
@@ -84,6 +157,248 @@ OPENAPI_TAGS = [
},
]
_JSON_SCHEMA_REF_PREFIX = "#/components/schemas/"
_OPENAPI_EXAMPLE_TAGS = {
"agents",
"activity",
"gateways",
"metrics",
"organizations",
"souls-directory",
"skills",
"board-groups",
"board-group-memory",
"boards",
"board-memory",
"board-webhooks",
"board-onboarding",
"approvals",
"tasks",
"custom-fields",
"tags",
"users",
}
_GENERIC_RESPONSE_DESCRIPTIONS = {"Successful Response", "Validation Error"}
_HTTP_RESPONSE_DESCRIPTIONS = {
"200": "Request completed successfully.",
"201": "Resource created successfully.",
"202": "Request accepted for processing.",
"204": "Request completed successfully with no response body.",
"400": "Request validation failed.",
"401": "Authentication is required or token is invalid.",
"403": "Caller is authenticated but not authorized for this operation.",
"404": "Requested resource was not found.",
"409": "Request conflicts with the current resource state.",
"422": "Request payload failed schema or field validation.",
"429": "Request was rate-limited.",
"500": "Internal server error.",
}
_METHOD_SUMMARY_PREFIX = {
"get": "List",
"post": "Create",
"put": "Replace",
"patch": "Update",
"delete": "Delete",
}
def _resolve_schema_ref(
schema: dict[str, Any],
*,
components: dict[str, Any],
seen_refs: set[str] | None = None,
) -> dict[str, Any]:
"""Resolve local component refs for OpenAPI schema traversal."""
ref = schema.get("$ref")
if not isinstance(ref, str):
return schema
if not ref.startswith(_JSON_SCHEMA_REF_PREFIX):
return schema
if seen_refs is None:
seen_refs = set()
if ref in seen_refs:
return schema
seen_refs.add(ref)
schema_name = ref[len(_JSON_SCHEMA_REF_PREFIX) :]
schemas = components.get("schemas")
if not isinstance(schemas, dict):
return schema
target = schemas.get(schema_name)
if not isinstance(target, dict):
return schema
return _resolve_schema_ref(target, components=components, seen_refs=seen_refs)
def _example_from_schema(schema: dict[str, Any], *, components: dict[str, Any]) -> Any:
"""Generate an OpenAPI example from schema metadata with sensible fallbacks."""
resolved = _resolve_schema_ref(schema, components=components)
if "example" in resolved:
return resolved["example"]
examples = resolved.get("examples")
if isinstance(examples, list) and examples:
return examples[0]
for composite_key in ("anyOf", "oneOf", "allOf"):
composite = resolved.get(composite_key)
if isinstance(composite, list):
for branch in composite:
if not isinstance(branch, dict):
continue
branch_example = _example_from_schema(branch, components=components)
if branch_example is not None:
return branch_example
enum_values = resolved.get("enum")
if isinstance(enum_values, list) and enum_values:
return enum_values[0]
schema_type = resolved.get("type")
if schema_type == "object":
output: dict[str, Any] = {}
properties = resolved.get("properties")
if isinstance(properties, dict):
for key, property_schema in properties.items():
if not isinstance(property_schema, dict):
continue
property_example = _example_from_schema(property_schema, components=components)
if property_example is not None:
output[key] = property_example
if output:
return output
additional_properties = resolved.get("additionalProperties")
if isinstance(additional_properties, dict):
value_example = _example_from_schema(additional_properties, components=components)
if value_example is not None:
return {"key": value_example}
return {}
if schema_type == "array":
items = resolved.get("items")
if isinstance(items, dict):
item_example = _example_from_schema(items, components=components)
if item_example is not None:
return [item_example]
return []
if schema_type == "string":
return "string"
if schema_type == "integer":
return 0
if schema_type == "number":
return 0
if schema_type == "boolean":
return False
return None
def _inject_json_content_example(
*,
content: dict[str, Any],
components: dict[str, Any],
) -> None:
"""Attach an example to application/json content when one is missing."""
app_json = content.get("application/json")
if not isinstance(app_json, dict):
return
if "example" in app_json or "examples" in app_json:
return
schema = app_json.get("schema")
if not isinstance(schema, dict):
return
generated_example = _example_from_schema(schema, components=components)
if generated_example is not None:
app_json["example"] = generated_example
def _build_operation_summary(*, method: str, path: str) -> str:
"""Build a readable summary when an operation does not define one."""
prefix = _METHOD_SUMMARY_PREFIX.get(method.lower(), "Handle")
path_without_prefix = path.removeprefix("/api/v1/")
parts = [
part.replace("-", " ")
for part in path_without_prefix.split("/")
if part and not (part.startswith("{") and part.endswith("}"))
]
if not parts:
return prefix
return f"{prefix} {' '.join(parts)}".strip().title()
def _normalize_operation_docs(
*,
operation: dict[str, Any],
method: str,
path: str,
) -> None:
"""Normalize summary/description/responses/request-body docs for tagged operations."""
summary = str(operation.get("summary", "")).strip()
if not summary:
summary = _build_operation_summary(method=method, path=path)
operation["summary"] = summary
description = str(operation.get("description", "")).strip()
if not description:
operation["description"] = f"{summary}."
request_body = operation.get("requestBody")
if isinstance(request_body, dict):
if not str(request_body.get("description", "")).strip():
request_body["description"] = "JSON request payload."
responses = operation.get("responses")
if not isinstance(responses, dict):
return
for status_code, response in responses.items():
if not isinstance(response, dict):
continue
existing_description = str(response.get("description", "")).strip()
if not existing_description or existing_description in _GENERIC_RESPONSE_DESCRIPTIONS:
response["description"] = _HTTP_RESPONSE_DESCRIPTIONS.get(
str(status_code),
"Request processed.",
)
def _inject_tagged_operation_openapi_docs(openapi_schema: dict[str, Any]) -> None:
"""Ensure targeted-tag operations expose consistent OpenAPI docs and examples."""
components = openapi_schema.get("components")
if not isinstance(components, dict):
return
paths = openapi_schema.get("paths")
if not isinstance(paths, dict):
return
for path, path_item in paths.items():
if not isinstance(path_item, dict):
continue
for method, operation in path_item.items():
if not isinstance(operation, dict):
continue
tags = operation.get("tags")
if not isinstance(tags, list):
continue
if not _OPENAPI_EXAMPLE_TAGS.intersection(tags):
continue
_normalize_operation_docs(operation=operation, method=method, path=path)
request_body = operation.get("requestBody")
if isinstance(request_body, dict):
request_content = request_body.get("content")
if isinstance(request_content, dict):
_inject_json_content_example(content=request_content, components=components)
responses = operation.get("responses")
if isinstance(responses, dict):
for response in responses.values():
if not isinstance(response, dict):
continue
response_content = response.get("content")
if isinstance(response_content, dict):
_inject_json_content_example(content=response_content, components=components)
@asynccontextmanager
async def lifespan(_: FastAPI) -> AsyncIterator[None]:
@@ -203,5 +518,26 @@ api_v1.include_router(tags_router)
api_v1.include_router(users_router)
app.include_router(api_v1)
def custom_openapi() -> dict[str, Any]:
"""Generate OpenAPI schema with normalized docs/examples for targeted tags."""
if app.openapi_schema:
return app.openapi_schema
openapi_schema = get_openapi(
title=app.title,
version=app.version,
openapi_version=app.openapi_version,
description=app.description,
routes=app.routes,
tags=app.openapi_tags,
servers=app.servers,
)
_inject_tagged_operation_openapi_docs(openapi_schema)
app.openapi_schema = openapi_schema
return app.openapi_schema
app.openapi = custom_openapi
add_pagination(app)
logger.debug("app.routes.registered count=%s", len(app.routes))