Add utility script to export openapi.json

pyproject.toml

@@ -8,7 +8,9 @@ dev = [
   "pytest>=8.4.0",
   "pytest-asyncio>=1.0.0",
   "pytest-cov>=6.1.1",
-  "httpx>=0.25.0"
+  "httpx>=0.25.0",
+  "hatch>=1.14.1",
+  "pyyaml>=6.0.2"
 ]
 
 [project]
@@ -36,3 +38,9 @@ omit = ["tests/*"]
 
 [tool.hatch.build.targets.wheel]
 packages = ["src/workflows"]
+
+[tool.hatch.envs.server]
+features = ["server"]
+
+[tool.hatch.envs.server.scripts]
+openapi = "python -m workflows.server.server --output openapi.json"

src/workflows/server/server.py

@@ -15,12 +15,13 @@
 from starlette.requests import Request
 from starlette.responses import JSONResponse, StreamingResponse
 from starlette.routing import Route
+from starlette.schemas import SchemaGenerator
 
 from workflows import Context, Workflow
 from workflows.context.serializers import JsonSerializer
 from workflows.events import StopEvent
 from workflows.handler import WorkflowHandler
-
+from importlib.metadata import version
 from .utils import nanoid
 
 logger = logging.getLogger()
@@ -435,3 +436,33 @@ async def _extract_run_params(self, request: Request, workflow: Workflow) -> tup
             raise HTTPException(
                 detail=f"Error processing request body: {e}", status_code=500
             )
+
+    def openapi_schema(self) -> dict:
+        app = self.app
+        gen = SchemaGenerator(
+            {
+                "openapi": "3.0.0",
+                "info": {
+                    "title": "Workflows API",
+                    "version": version("llama-index-workflows"),
+                },
+            }
+        )
+
+        return gen.get_schema(app.routes)
+
+
+if __name__ == "__main__":
+    import argparse
+
+    parser = argparse.ArgumentParser(description="Generate OpenAPI schema")
+    parser.add_argument(
+        "--output", type=str, default="openapi.json", help="Output file path"
+    )
+    args = parser.parse_args()
+
+    server = WorkflowServer()
+    dict_schema = server.openapi_schema()
+    with open(args.output, "w") as f:
+        json.dump(dict_schema, indent=2, fp=f)
+    print(f"OpenAPI schema written to {args.output}")

tests/server/test_openapi_schema.py

@@ -1,7 +1,6 @@
 # SPDX-License-Identifier: MIT
 # Copyright (c) 2025 LlamaIndex Inc.
 
-from starlette.schemas import SchemaGenerator
 from workflows import Workflow
 
 from workflows.server import WorkflowServer
@@ -10,19 +9,7 @@
 def test_openapi_schema_includes_all_routes(simple_test_workflow: Workflow) -> None:
     server = WorkflowServer()
     server.add_workflow("test", simple_test_workflow)
-
-    app = server.app
-    gen = SchemaGenerator(
-        {
-            "openapi": "3.0.0",
-            "info": {
-                "title": "Workflows API",
-                "version": "1.0.0",
-            },
-        }
-    )
-
-    schema = gen.get_schema(app.routes)
+    schema = server.openapi_schema()
 
     assert "paths" in schema
     paths = schema["paths"]