[WIP - not draft] feat: synchronize-openapi-schema-with-gram

.github/workflows/api-deploy-production-ecs.yml

@@ -1,5 +1,8 @@
 name: API Deploy to Production ECS
 
+permissions:
+  contents: read
+
 on:
   push:
     tags:
@@ -15,3 +18,42 @@ jobs:
     with:
       environment: production
     secrets: inherit
+
+  mcp-schema-push:
+    name: Push MCP Schema to Gram
+    runs-on: depot-ubuntu-latest
+    defaults:
+      run:
+        working-directory: api
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Install Poetry
+        run: make install-poetry
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+          cache: poetry
+
+      - name: Install dependencies
+        run: |
+          echo "https://${{ secrets.GH_PRIVATE_ACCESS_TOKEN }}:@github.com" > ${HOME}/.git-credentials
+          git config --global credential.helper store
+          make install-packages opts="--with saml,auth-controller,workflows,release-pipelines"
+          make install-private-modules
+          rm -rf ${HOME}/.git-credentials
+
+      - name: Generate MCP schema
+        run: make generate-mcp-spec
+
+      - name: Install Gram CLI
+        run: curl -fsSL https://go.getgram.ai/cli.sh | bash
+
+      - name: Push to Gram
+        env:
+          GRAM_API_KEY: ${{ secrets.GRAM_API_KEY }}
+          GRAM_ORG: ${{ secrets.GRAM_ORG }}
+          GRAM_PROJECT: ${{ secrets.GRAM_PROJECT }}
+        run: gram push --api-key "$GRAM_API_KEY" --org "$GRAM_ORG" --project "$GRAM_PROJECT" --config gram.json

api/Makefile

@@ -156,3 +156,7 @@ generate-docs:
 .PHONY: add-known-sdk-version
 add-known-sdk-version:
 	poetry run python scripts/add-known-sdk-version.py $(opts)
+
+.PHONY: generate-mcp-spec
+generate-mcp-spec:
+	poetry run python manage.py spectacular --generator-class api.openapi.MCPSchemaGenerator --file mcp_openapi.yaml

api/api/openapi.py

@@ -51,7 +51,91 @@ def get_schema(
         self, request: Request | None = None, public: bool = False
     ) -> dict[str, Any]:
         schema: dict[str, Any] = super().get_schema(request, public)  # type: ignore[no-untyped-call]
-        schema["$schema"] = "https://spec.openapis.org/oas/3.1/dialect/base"
+        return {
+            "$schema": "https://spec.openapis.org/oas/3.1/dialect/base",
+            **schema,
+        }
+
+
+class MCPSchemaGenerator(SchemaGenerator):
+    """
+    Schema generator that filters to only include operations tagged with "mcp".
+
+    Supports custom extensions:
+    - x-mcp-name: Override the operationId for MCP tools
+    - x-mcp-description: Override the description for MCP tools
+    """
+
+    MCP_TAG = "mcp"
+    MCP_SERVER_URL = "https://api.flagsmith.com"
+
+    def get_schema(
+        self, request: Request | None = None, public: bool = False
+    ) -> dict[str, Any]:
+        schema = super().get_schema(request, public)
+        schema["paths"] = self._filter_paths(schema.get("paths", {}))
+        schema = self._update_security_for_mcp(schema)
+        schema.pop("$schema", None)
+        info = schema.pop("info")
+        info["title"] = "mcp_openapi"
+        return {
+            "openapi": schema.pop("openapi"),
+            "info": info,
+            "servers": [{"url": self.MCP_SERVER_URL}],
+            **schema,
+        }
+
+    def _filter_paths(self, paths: dict[str, Any]) -> dict[str, Any]:
+        """Filter paths to only include operations tagged with 'mcp'."""
+        filtered_paths: dict[str, Any] = {}
+
+        for path, path_item in paths.items():
+            filtered_operations: dict[str, Any] = {}
+
+            for method, operation in path_item.items():
+                if not isinstance(operation, dict):
+                    filtered_operations[method] = operation
+                    continue
+
+                tags = operation.get("tags", [])
+                if self.MCP_TAG in tags:
+                    filtered_operations[method] = self._transform_for_mcp(operation)
+
+            if any(isinstance(op, dict) for op in filtered_operations.values()):
+                filtered_paths[path] = filtered_operations
+
+        return filtered_paths
+
+    def _transform_for_mcp(self, operation: dict[str, Any]) -> dict[str, Any]:
+        """Apply MCP-specific transformations to an operation."""
+        operation = operation.copy()
+
+        # Override operationId if x-mcp-name provided
+        if mcp_name := operation.pop("x-mcp-name", None):
+            operation["operationId"] = mcp_name
+
+        # Override description if x-mcp-description provided
+        if mcp_desc := operation.pop("x-mcp-description", None):
+            operation["description"] = mcp_desc
+
+        # Remove operation-level security (use global MCP security instead)
+        operation.pop("security", None)
+
+        return operation
+
+    def _update_security_for_mcp(self, schema: dict[str, Any]) -> dict[str, Any]:
+        """Update security schemes for MCP (Organisation API Key)."""
+        schema = schema.copy()
+        schema["components"] = schema.get("components", {}).copy()
+        schema["components"]["securitySchemes"] = {
+            "TOKEN_AUTH": {
+                "type": "apiKey",
+                "in": "header",
+                "name": "Authorization",
+                "description": "Organisation API Key. Format: Api-Key <key>",
+            },
+        }
+        schema["security"] = [{"TOKEN_AUTH": []}]
         return schema
 
 

api/api/openapi_views.py

@@ -0,0 +1,36 @@
+from typing import Any
+
+from drf_spectacular.views import SpectacularJSONAPIView, SpectacularYAMLAPIView
+from rest_framework.request import Request
+from rest_framework.response import Response
+
+from api.openapi import MCPSchemaGenerator, SchemaGenerator
+
+
+class CustomSpectacularJSONAPIView(SpectacularJSONAPIView):  # type: ignore[misc]
+    """
+    JSON schema view that supports ?mcp=true query parameter for MCP-filtered output.
+    """
+
+    def get(self, request: Request, *args: Any, **kwargs: Any) -> Response:
+        if request.query_params.get("mcp", "").lower() == "true":
+            self.generator_class = MCPSchemaGenerator
+        else:
+            self.generator_class = SchemaGenerator
+        return super().get(request, *args, **kwargs)
+
+
+class CustomSpectacularYAMLAPIView(SpectacularYAMLAPIView):  # type: ignore[misc]
+    """
+    YAML schema view that supports ?mcp=true query parameter for MCP-filtered output.
+    """
+
+    def get(self, request: Request, *args: Any, **kwargs: Any) -> Response:
+        if request.query_params.get("mcp", "").lower() == "true":
+            self.generator_class = MCPSchemaGenerator
+            response = super().get(request, *args, **kwargs)
+            response["Content-Disposition"] = 'attachment; filename="mcp_openapi.yaml"'
+            return response
+        else:
+            self.generator_class = SchemaGenerator
+            return super().get(request, *args, **kwargs)

api/api/urls/v1.py

@@ -1,12 +1,9 @@
 from django.conf import settings
 from django.urls import include, path, re_path
-from drf_spectacular.views import (
-    SpectacularJSONAPIView,
-    SpectacularSwaggerView,
-    SpectacularYAMLAPIView,
-)
+from drf_spectacular.views import SpectacularSwaggerView
 from rest_framework import permissions, routers
 
+from api.openapi_views import CustomSpectacularJSONAPIView, CustomSpectacularYAMLAPIView
 from app_analytics.views import SDKAnalyticsFlags, SelfHostedTelemetryAPIView
 from environments.identities.traits.views import SDKTraits
 from environments.identities.views import SDKIdentities
@@ -73,14 +70,14 @@
     # API documentation
     path(
         "swagger.json",
-        SpectacularJSONAPIView.as_view(
+        CustomSpectacularJSONAPIView.as_view(
             permission_classes=[schema_view_permission_class],
         ),
         name="schema-json",
     ),
     path(
         "swagger.yaml",
-        SpectacularYAMLAPIView.as_view(
+        CustomSpectacularYAMLAPIView.as_view(
             permission_classes=[schema_view_permission_class],
         ),
         name="schema-yaml",

api/environments/views.py

@@ -69,6 +69,7 @@
 @method_decorator(
     name="list",
     decorator=extend_schema(
+        tags=["mcp"],
         parameters=[
             OpenApiParameter(
                 name="project",
@@ -77,7 +78,11 @@
                 required=False,
                 type=int,
             )
-        ]
+        ],
+        extensions={
+            "x-mcp-name": "list_environments",
+            "x-mcp-description": "Lists all environments the user has access to.",
+        },
     ),
 )
 class EnvironmentViewSet(viewsets.ModelViewSet):  # type: ignore[type-arg]

api/features/feature_external_resources/views.py

@@ -1,6 +1,8 @@
 import re
 
 from django.shortcuts import get_object_or_404
+from django.utils.decorators import method_decorator
+from drf_spectacular.utils import extend_schema
 from rest_framework import status, viewsets
 from rest_framework.response import Response
 
@@ -17,6 +19,16 @@
 from .serializers import FeatureExternalResourceSerializer
 
 
+@method_decorator(
+    name="list",
+    decorator=extend_schema(
+        tags=["mcp"],
+        extensions={
+            "x-mcp-name": "get_feature_external_resources",
+            "x-mcp-description": "Retrieves external resources linked to the feature flag.",
+        },
+    ),
+)
 class FeatureExternalResourceViewSet(viewsets.ModelViewSet):  # type: ignore[type-arg]
     serializer_class = FeatureExternalResourceSerializer
     permission_classes = [FeatureExternalResourcePermissions]

api/features/feature_health/views.py

@@ -6,6 +6,7 @@
 )
 from django.db.models import QuerySet
 from django.shortcuts import get_object_or_404
+from django.utils.decorators import method_decorator
 from drf_spectacular.utils import extend_schema
 from rest_framework import mixins, status, viewsets
 from rest_framework.decorators import action, api_view, permission_classes
@@ -34,6 +35,16 @@
 from users.models import FFAdminUser
 
 
+@method_decorator(
+    name="list",
+    decorator=extend_schema(
+        tags=["mcp"],
+        extensions={
+            "x-mcp-name": "get_feature_health_events",
+            "x-mcp-description": "Retrieves feature health monitoring events and metrics for the project.",
+        },
+    ),
+)
 class FeatureHealthEventViewSet(
     mixins.ListModelMixin,
     viewsets.GenericViewSet[FeatureHealthEvent],

api/features/multivariate/views.py

@@ -2,6 +2,7 @@
     CREATE_FEATURE,
     VIEW_PROJECT,
 )
+from django.utils.decorators import method_decorator
 from drf_spectacular.utils import extend_schema
 from rest_framework import viewsets
 from rest_framework.decorators import api_view
@@ -15,6 +16,46 @@
 from .serializers import MultivariateFeatureOptionSerializer
 
 
+@method_decorator(
+    name="list",
+    decorator=extend_schema(
+        tags=["mcp"],
+        extensions={
+            "x-mcp-name": "list_feature_multivariate_options",
+            "x-mcp-description": "Retrieves all multivariate options for a feature flag.",
+        },
+    ),
+)
+@method_decorator(
+    name="create",
+    decorator=extend_schema(
+        tags=["mcp"],
+        extensions={
+            "x-mcp-name": "create_feature_multivariate_option",
+            "x-mcp-description": "Creates a new multivariate option for a feature flag.",
+        },
+    ),
+)
+@method_decorator(
+    name="update",
+    decorator=extend_schema(
+        tags=["mcp"],
+        extensions={
+            "x-mcp-name": "update_feature_multivariate_option",
+            "x-mcp-description": "Updates an existing multivariate option.",
+        },
+    ),
+)
+@method_decorator(
+    name="destroy",
+    decorator=extend_schema(
+        tags=["mcp"],
+        extensions={
+            "x-mcp-name": "delete_feature_multivariate_option",
+            "x-mcp-description": "Deletes a multivariate option.",
+        },
+    ),
+)
 class MultivariateFeatureOptionViewSet(viewsets.ModelViewSet):  # type: ignore[type-arg]
     serializer_class = MultivariateFeatureOptionSerializer