From 9b1b563a354c710b8ee05c15dc004d7abd31f9e0 Mon Sep 17 00:00:00 2001 From: Ben Date: Fri, 26 Dec 2025 20:17:00 +0000 Subject: [PATCH] fix: accept both str and dict params in komodo_api_call Resolves issue #1 - allows params parameter to accept either JSON string or dict type, fixing type validation errors when calling the function. --- server.py | 64 ++++++++++++++++++++++++++++++++----------------------- 1 file changed, 37 insertions(+), 27 deletions(-) diff --git a/server.py b/server.py index f7ef859..1a02aca 100644 --- a/server.py +++ b/server.py @@ -1,6 +1,7 @@ import os import json import logging +from typing import Any import httpx from fastmcp import FastMCP from starlette.applications import Starlette @@ -23,7 +24,7 @@ KOMODO_API_SECRET = os.getenv("KOMODO_API_SECRET", "") class KomodoClient: """HTTP client for Komodo Core API.""" - + def __init__(self, url: str, api_key: str, api_secret: str): self.url = url self.headers = { @@ -32,10 +33,10 @@ class KomodoClient: "X-Api-Secret": api_secret, } self._client = httpx.Client(timeout=30.0) - + def _request(self, endpoint: str, request_type: str, params: dict = None) -> dict: """Make a request to Komodo API. - + Args: endpoint: API endpoint (read, write, execute) request_type: Request type name (e.g., 'ListServers', 'GetDeployment') @@ -43,7 +44,7 @@ class KomodoClient: """ url = f"{self.url}/{endpoint}" body = {"type": request_type, "params": params or {}} - + try: response = self._client.post(url, json=body, headers=self.headers) response.raise_for_status() @@ -52,15 +53,15 @@ class KomodoClient: return {"error": f"HTTP {e.response.status_code}: {e.response.text}"} except Exception as e: return {"error": str(e)} - + def read(self, request_type: str, params: dict = None) -> dict: """Make a read request.""" return self._request("read", request_type, params) - + def write(self, request_type: str, params: dict = None) -> dict: """Make a write request.""" return self._request("write", request_type, params) - + def execute(self, request_type: str, params: dict = None) -> dict: """Make an execute request.""" return self._request("execute", request_type, params) @@ -145,10 +146,11 @@ def get_api_reference() -> str: # Note: Return str (JSON) instead of dict to work around FastMCP/Gemini CLI # schema compatibility issue with additionalProperties + @mcp.tool() def list_servers() -> str: """Lists all servers (periphery nodes) connected to Komodo. - + Returns server names, IDs, status, and system information as JSON. """ client, error = _get_client() @@ -160,7 +162,7 @@ def list_servers() -> str: @mcp.tool() def list_deployments() -> str: """Lists all deployments (single container applications). - + Deployments are Docker containers managed by Komodo. Returns JSON. """ client, error = _get_client() @@ -172,7 +174,7 @@ def list_deployments() -> str: @mcp.tool() def list_stacks() -> str: """Lists all stacks (docker-compose applications). - + Stacks are multi-container applications defined by docker-compose files. Returns JSON. """ client, error = _get_client() @@ -184,10 +186,10 @@ def list_stacks() -> str: @mcp.tool() def get_container_status(deployment: str) -> str: """Gets the status of a specific container. - + Args: deployment: Deployment name or ID - + Returns container state, image, ports, and resource usage as JSON. """ client, error = _get_client() @@ -198,19 +200,17 @@ def get_container_status(deployment: str) -> str: @mcp.tool() def komodo_api_call( - endpoint: str, - request_type: str, - params: str = "{}" + endpoint: str, request_type: str, params: str | dict[str, Any] = "{}" ) -> str: """Execute a raw Komodo API call. - + Use the komodo://api-reference resource for available request types. - + Args: endpoint: API endpoint - 'read', 'write', or 'execute' request_type: Request type name (e.g., 'GetServer', 'Deploy', 'ListBuilds') - params: Request parameters as a JSON string (default: "{}") - + params: Request parameters as a JSON string or dict (default: "{}") + Examples: - Get server: endpoint='read', request_type='GetServer', params='{"server": "my-server"}' - Deploy: endpoint='execute', request_type='Deploy', params='{"deployment": "my-app"}' @@ -219,15 +219,22 @@ def komodo_api_call( client, error = _get_client() if error: return json.dumps(error) - + if endpoint not in ("read", "write", "execute"): - return json.dumps({"error": f"Invalid endpoint '{endpoint}'. Use 'read', 'write', or 'execute'."}) - + return json.dumps( + { + "error": f"Invalid endpoint '{endpoint}'. Use 'read', 'write', or 'execute'." + } + ) + try: - parsed_params = json.loads(params) if params else {} + if isinstance(params, str): + parsed_params = json.loads(params) if params else {} + else: + parsed_params = params or {} except json.JSONDecodeError as e: return json.dumps({"error": f"Invalid JSON in params: {e}"}) - + method = getattr(client, endpoint) return json.dumps(method(request_type, parsed_params)) @@ -237,7 +244,9 @@ async def health(request): """Health check endpoint for Docker.""" client, error = _get_client() if error: - return JSONResponse({"status": "degraded", "error": error["error"]}, status_code=503) + return JSONResponse( + {"status": "degraded", "error": error["error"]}, status_code=503 + ) return JSONResponse({"status": "ok"}) @@ -245,14 +254,14 @@ async def health(request): def create_app(): """Create the ASGI application with health check and MCP routes.""" mcp_app = mcp.http_app() - + # Wrapper app: /health is standalone, everything else goes to MCP # IMPORTANT: Must pass mcp_app.lifespan for task group initialization routes = [ Route("/health", health, methods=["GET"]), Mount("/", app=mcp_app), # MCP handles /mcp endpoint ] - + return Starlette(routes=routes, lifespan=mcp_app.lifespan) @@ -262,4 +271,5 @@ app = create_app() if __name__ == "__main__": import uvicorn + uvicorn.run(app, host="0.0.0.0", port=8000)