feat: add lint checks

Signed-off-by: Eloy Coto <eloy.coto@acalustra.com>

Author: eloycoto

.github/workflows/ci.yml

@@ -0,0 +1,31 @@
+name: CI
+
+on:
+  push:
+    branches: [ master, test ]
+  pull_request:
+    branches: [ master ]
+
+jobs:
+  lint-and-test:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.13"]
+
+    steps:
+    - uses: actions/checkout@v5
+
+    - name: Install uv
+      uses: astral-sh/setup-uv@v3
+      with:
+        version: "latest"
+
+    - name: Install dependencies
+      run: |
+        uv sync --dev
+
+    - name: Lint with ruff
+      run: |
+        uv run ruff check .
+        uv run ruff format --check .

Makefile

@@ -2,6 +2,8 @@
 
 help:
 	@echo "Available targets:"
+	@echo "  check 		   	- Run ruff check for lint issue"
+	@echo "  format    		- Run ruff format"
 	@echo "  mcp-run    	- Run MCP server locally with uv"
 	@echo "  run-servers    - Run llamastack and lightspeed in a single command for develop"
 
@@ -10,3 +12,10 @@ mcp-run:
 
 run-servers:
 	podman-compose up llama-stack lightspeed-stack
+
+check:
+	uv run ruff check . --fix
+
+format:
+	uv run ruff format
+

mcp_server.py

@@ -1,24 +1,23 @@
-from fastmcp import FastMCP
 import logging
+
+from fastmcp import FastMCP
+
 from tools.orchestrator_service import orchestrator_mcp
-import tools.get_orchestrator_instances
-import tools.orchestrator_creation_workflow_rules
 
 # Set up logging to see what's happening
-logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+logging.basicConfig(
+    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
+)
 logger = logging.getLogger(__name__)
 
-mcp = FastMCP(
-    name="Current Date and Time", port=8000
-)
+mcp = FastMCP(name="Current Date and Time", port=8000)
 
 # Mount the orchestrator service
 mcp.mount(orchestrator_mcp, prefix="orchestrator")
 
 
-
 if __name__ == "__main__":
     mcp.run(
-            transport="http",
-            host="0.0.0.0",  
-        )
+        transport="http",
+        host="0.0.0.0",
+    )

pyproject.toml

@@ -38,3 +38,8 @@ ignore = []
 [tool.ruff.format]
 quote-style = "double"
 indent-style = "space"
+
+[dependency-groups]
+dev = [
+    "ruff>=0.12.9",
+]

serverless-workflow/consolidate_schemas.py

@@ -7,9 +7,8 @@
 
 import json
 import os
-import re
-from typing import Dict, Any, Set
 from pathlib import Path
+from typing import Any, Dict, Set
 
 
 class SchemaConsolidator:
@@ -24,7 +23,7 @@ def load_schema(self, filename: str) -> Dict[str, Any]:
         if filename not in self.schemas:
             file_path = self.schema_dir / filename
             try:
-                with open(file_path, 'r') as f:
+                with open(file_path, "r") as f:
                     self.schemas[filename] = json.load(f)
                 print(f"Loaded schema: {filename}")
             except FileNotFoundError:
@@ -34,26 +33,30 @@ def load_schema(self, filename: str) -> Dict[str, Any]:
 
     def extract_ref_filename(self, ref: str) -> str:
         """Extract filename from $ref string."""
-        # Handle refs like "secrets.json#/secrets" or "common.json#/definitions/metadata"
-        if '#' in ref:
-            return ref.split('#')[0]
+        # Handle refs like "secrets.json#/secrets" or
+        # "common.json#/definitions/metadata"
+        if "#" in ref:
+            return ref.split("#")[0]
         return ref
 
     def extract_ref_path(self, ref: str) -> str:
         """Extract JSON path from $ref string."""
-        # Handle refs like "secrets.json#/secrets" or "common.json#/definitions/metadata"
-        if '#' in ref:
-            return ref.split('#')[1]
+        # Handle refs like "secrets.json#/secrets" or
+        # "common.json#/definitions/metadata"
+        if "#" in ref:
+            return ref.split("#")[1]
         return ""
 
-    def resolve_external_refs(self, schema: Dict[str, Any], current_file: str = "") -> Dict[str, Any]:
+    def resolve_external_refs(
+        self, schema: Dict[str, Any], current_file: str = ""
+    ) -> Dict[str, Any]:
         """Recursively resolve external $ref references."""
         if isinstance(schema, dict):
-            if '$ref' in schema:
-                ref = schema['$ref']
+            if "$ref" in schema:
+                ref = schema["$ref"]
 
                 # Skip internal references (starting with #)
-                if ref.startswith('#'):
+                if ref.startswith("#"):
                     return schema
 
                 # Extract filename and path
@@ -68,12 +71,15 @@ def resolve_external_refs(self, schema: Dict[str, Any], current_file: str = "")
                         # Navigate to the specific path in the referenced schema
                         target = ref_schema
                         if ref_path:
-                            path_parts = ref_path.strip('/').split('/')
+                            path_parts = ref_path.strip("/").split("/")
                             for part in path_parts:
                                 if part in target:
                                     target = target[part]
                                 else:
-                                    print(f"Warning: Path {ref_path} not found in {ref_file}")
+                                    print(
+                                        f"Warning: Path {ref_path} not found in "
+                                        f"{ref_file}"
+                                    )
                                     return schema
 
                         # Recursively resolve refs in the target
@@ -100,25 +106,31 @@ def resolve_external_refs(self, schema: Dict[str, Any], current_file: str = "")
 
     def process_schema_definitions(self, schema: Dict[str, Any], filename: str):
         """Extract and merge definitions from a schema into consolidated definitions."""
-        if 'definitions' in schema:
-            for def_name, def_value in schema['definitions'].items():
+        if "definitions" in schema:
+            for def_name, def_value in schema["definitions"].items():
                 # Create unique key to avoid conflicts
                 unique_key = f"{filename.replace('.json', '')}_{def_name}"
-                self.consolidated_definitions[unique_key] = self.resolve_external_refs(def_value, filename)
+                self.consolidated_definitions[unique_key] = self.resolve_external_refs(
+                    def_value, filename
+                )
                 print(f"Added definition: {unique_key}")
 
-    def update_internal_refs(self, schema: Dict[str, Any], filename: str) -> Dict[str, Any]:
+    def update_internal_refs(
+        self, schema: Dict[str, Any], filename: str
+    ) -> Dict[str, Any]:
         """Update internal references to point to consolidated definitions."""
         if isinstance(schema, dict):
-            if '$ref' in schema:
-                ref = schema['$ref']
+            if "$ref" in schema:
+                ref = schema["$ref"]
 
                 # Handle internal references
-                if ref.startswith('#/definitions/'):
-                    def_name = ref.replace('#/definitions/', '')
+                if ref.startswith("#/definitions/"):
+                    def_name = ref.replace("#/definitions/", "")
                     # Update to use the prefixed definition name
-                    new_ref = f"#/definitions/{filename.replace('.json', '')}_{def_name}"
-                    return {'$ref': new_ref}
+                    new_ref = (
+                        f"#/definitions/{filename.replace('.json', '')}_{def_name}"
+                    )
+                    return {"$ref": new_ref}
 
                 return schema
             else:
@@ -146,7 +158,7 @@ def consolidate(self, main_schema_file: str = "workflow.json") -> Dict[str, Any]
         # First pass: collect all definitions from all referenced schemas
         print("\n--- Collecting definitions from all schemas ---")
         for filename in os.listdir(self.schema_dir):
-            if filename.endswith('.json'):
+            if filename.endswith(".json"):
                 schema = self.load_schema(filename)
                 self.process_schema_definitions(schema, filename)
 
@@ -159,20 +171,22 @@ def consolidate(self, main_schema_file: str = "workflow.json") -> Dict[str, Any]
         final_schema = self.update_internal_refs(resolved_main, main_schema_file)
 
         # Add all consolidated definitions
-        if 'definitions' not in final_schema:
-            final_schema['definitions'] = {}
+        if "definitions" not in final_schema:
+            final_schema["definitions"] = {}
 
-        final_schema['definitions'].update(self.consolidated_definitions)
+        final_schema["definitions"].update(self.consolidated_definitions)
 
-        print(f"\n--- Consolidation complete ---")
+        print("\n--- Consolidation complete ---")
         print(f"Total definitions consolidated: {len(self.consolidated_definitions)}")
 
         return final_schema
 
-    def save_consolidated_schema(self, consolidated_schema: Dict[str, Any], output_file: str):
+    def save_consolidated_schema(
+        self, consolidated_schema: Dict[str, Any], output_file: str
+    ):
         """Save the consolidated schema to a file."""
         output_path = output_file
-        with open(output_path, 'w') as f:
+        with open(output_path, "w") as f:
             json.dump(consolidated_schema, f, indent=2)
         print(f"Consolidated schema saved to: {output_path}")
 
@@ -191,7 +205,9 @@ def main():
         consolidated = consolidator.consolidate("workflow.json")
 
         # Save the result
-        consolidator.save_consolidated_schema(consolidated, "consolidated_workflow_schema.json")
+        consolidator.save_consolidated_schema(
+            consolidated, "consolidated_workflow_schema.json"
+        )
         print("\nConsolidation completed successfully!")
     except Exception as e:
         print(f"Error during consolidation: {e}")

tools/__init__.py

@@ -0,0 +1,3 @@
+from .orchestrator_creation_workflow_rules import creation_workflow_rules
+
+__all__ = [creation_workflow_rules]

tools/get_orchestrator_instances.py

@@ -1,6 +1,6 @@
-from .orchestrator_service import orchestrator_mcp
 import logging
-from typing import Dict, List, Any
+
+from .orchestrator_service import orchestrator_mcp
 
 logger = logging.getLogger(__name__)
 
@@ -299,7 +299,8 @@
 - Use `timeouts.eventTimeout` for event waiting limits (ISO 8601 duration format)
 - Consider error handling for event timeout scenarios
 - Events can carry data accessible in subsequent states via `.eventData`
-"""
+"""  # noqa: E501
+
 
 @orchestrator_mcp.tool()
 def creation_workflow_rules(session_id: str) -> str:

tools/orchestrator_creation_workflow_rules.py

@@ -1,6 +1,7 @@
-from fastmcp import FastMCP
 import logging
 
+from fastmcp import FastMCP
+
 logger = logging.getLogger(__name__)
 
-orchestrator_mcp = FastMCP("Orchestrator Service")
+orchestrator_mcp = FastMCP("Orchestrator Service")