Adjusting MCP header processing to llama stack 0.4.x and adding MCP e2e tests

Author: blublinsky

dev-tools/mcp-mock-server/README.md

@@ -19,9 +19,10 @@ This mock server helps developers:
 - ✅ **HTTP & HTTPS** - Runs both protocols simultaneously for comprehensive testing
 - ✅ **Header Capture** - Captures and displays all request headers
 - ✅ **Debug Endpoints** - Inspect captured headers and request history
-- ✅ **MCP Protocol** - Implements basic MCP endpoints for testing
+- ✅ **MCP Protocol** - Implements MCP endpoints (initialize, tools/list, tools/call)
 - ✅ **Request Logging** - Tracks recent requests with timestamps
 - ✅ **Self-Signed Certs** - Auto-generates certificates for HTTPS testing
+- ✅ **Tool Execution** - Returns mock results for tool/call testing
 
 ## Quick Start
 
@@ -46,8 +47,11 @@ HTTPS: https://localhost:3001
 Debug endpoints:
   • /debug/headers  - View captured headers
   • /debug/requests - View request log
-MCP endpoint:
-  • POST /mcp/v1/list_tools
+MCP endpoints:
+  • POST with JSON-RPC (any path)
+    - method: "initialize"
+    - method: "tools/list"
+    - method: "tools/call"
 ======================================================================
 Note: HTTPS uses a self-signed certificate (for testing only)
 ```
@@ -270,8 +274,9 @@ python dev-tools/mcp-mock-server/server.py 8080
 This is a **development/testing tool only**:
 - ❌ Not for production use
 - ❌ No authentication/security
-- ❌ Limited MCP protocol implementation
+- ❌ Limited MCP protocol implementation (initialize, tools/list, tools/call only)
 - ❌ Single-threaded (one request at a time)
+- ❌ Mock responses only (not real tool execution)
 
 For production, use real MCP servers.
 

dev-tools/mcp-mock-server/server.py

@@ -60,7 +60,11 @@ def _capture_headers(self) -> None:
         if len(request_log) > 10:
             request_log.pop(0)
 
-    def do_POST(self) -> None:  # pylint: disable=invalid-name
+    def do_POST(
+        self,
+    ) -> (
+        None
+    ):  # pylint: disable=invalid-name,too-many-locals,too-many-branches,too-many-statements
         """Handle POST requests (MCP protocol endpoints)."""
         self._capture_headers()
 
@@ -73,14 +77,24 @@ def do_POST(self) -> None:  # pylint: disable=invalid-name
             request_id = request_data.get("id", 1)
             method = request_data.get("method", "unknown")
         except (json.JSONDecodeError, UnicodeDecodeError):
+            request_data = {}
             request_id = 1
             method = "unknown"
 
+        # Log the RPC method in the request log
+        if request_log:
+            request_log[-1]["rpc_method"] = method
+
         # Determine tool name based on authorization header to avoid collisions
         auth_header = self.headers.get("Authorization", "")
 
+        # Initialize tool info defaults
+        tool_name = "mock_tool_no_auth"
+        tool_desc = "Mock tool with no authorization"
+        error_mode = False
+
         # Match based on token content
-        match auth_header:
+        match True:
             case _ if "test-secret-token" in auth_header:
                 tool_name = "mock_tool_file"
                 tool_desc = "Mock tool with file-based auth"
@@ -90,58 +104,116 @@ def do_POST(self) -> None:  # pylint: disable=invalid-name
             case _ if "my-client-token" in auth_header:
                 tool_name = "mock_tool_client"
                 tool_desc = "Mock tool with client-provided token"
+            case _ if "error-mode" in auth_header:
+                tool_name = "mock_tool_error"
+                tool_desc = "Mock tool configured to return errors"
+                error_mode = True
             case _:
-                # No auth header or unrecognized token
-                tool_name = "mock_tool_no_auth"
-                tool_desc = "Mock tool with no authorization"
-
-        # Handle MCP protocol methods
-        if method == "initialize":
-            # Return MCP initialize response
-            response = {
-                "jsonrpc": "2.0",
-                "id": request_id,
-                "result": {
-                    "protocolVersion": "2024-11-05",
-                    "capabilities": {
-                        "tools": {},
-                    },
-                    "serverInfo": {
-                        "name": "mock-mcp-server",
-                        "version": "1.0.0",
+                # Default case already set above
+                pass
+
+        # Log the tool name in the request log
+        if request_log:
+            request_log[-1]["tool_name"] = tool_name
+
+        # Handle MCP protocol methods using match statement
+        response: dict = {}
+        match method:
+            case "initialize":
+                # Return MCP initialize response
+                response = {
+                    "jsonrpc": "2.0",
+                    "id": request_id,
+                    "result": {
+                        "protocolVersion": "2024-11-05",
+                        "capabilities": {
+                            "tools": {},
+                        },
+                        "serverInfo": {
+                            "name": "mock-mcp-server",
+                            "version": "1.0.0",
+                        },
                     },
-                },
-            }
-        elif method == "tools/list":
-            # Return list of tools with unique name
-            response = {
-                "jsonrpc": "2.0",
-                "id": request_id,
-                "result": {
-                    "tools": [
-                        {
-                            "name": tool_name,
-                            "description": tool_desc,
-                            "inputSchema": {
-                                "type": "object",
-                                "properties": {
-                                    "message": {
-                                        "type": "string",
-                                        "description": "Test message",
-                                    }
+                }
+
+            case "tools/list":
+                # Return list of tools with unique name
+                response = {
+                    "jsonrpc": "2.0",
+                    "id": request_id,
+                    "result": {
+                        "tools": [
+                            {
+                                "name": tool_name,
+                                "description": tool_desc,
+                                "inputSchema": {
+                                    "type": "object",
+                                    "properties": {
+                                        "message": {
+                                            "type": "string",
+                                            "description": "Test message",
+                                        }
+                                    },
                                 },
-                            },
-                        }
-                    ]
-                },
-            }
-        else:
-            # Generic success response for other methods
-            response = {
-                "jsonrpc": "2.0",
-                "id": request_id,
-                "result": {"status": "ok"},
-            }
+                            }
+                        ]
+                    },
+                }
+
+            case "tools/call":
+                # Handle tool execution
+                params = request_data.get("params", {})
+                tool_called = params.get("name", "unknown")
+                arguments = params.get("arguments", {})
+
+                # Check if error mode is enabled
+                if error_mode:
+                    # Return error response
+                    response = {
+                        "jsonrpc": "2.0",
+                        "id": request_id,
+                        "result": {
+                            "content": [
+                                {
+                                    "type": "text",
+                                    "text": (
+                                        f"Error: Tool '{tool_called}' "
+                                        "execution failed - simulated error."
+                                    ),
+                                }
+                            ],
+                            "isError": True,
+                        },
+                    }
+                else:
+                    # Build result text
+                    result_text = (
+                        f"Mock tool '{tool_called}' executed successfully "
+                        f"with arguments: {arguments}."
+                    )
+
+                    # Return successful tool execution result
+                    response = {
+                        "jsonrpc": "2.0",
+                        "id": request_id,
+                        "result": {
+                            "content": [
+                                {
+                                    "type": "text",
+                                    "text": result_text,
+                                }
+                            ],
+                            "isError": False,
+                        },
+                    }
+
+            case _:
+                # Generic success response for other methods
+                response = {
+                    "jsonrpc": "2.0",
+                    "id": request_id,
+                    "result": {"status": "ok"},
+                }
 
         self.send_response(200)
         self.send_header("Content-Type", "application/json")
@@ -160,6 +232,11 @@ def do_GET(self) -> None:  # pylint: disable=invalid-name
                 )
             case "/debug/requests":
                 self._send_json_response(request_log)
+            case "/debug/clear":
+                # Clear the request log and last captured headers
+                request_log.clear()
+                last_headers.clear()
+                self._send_json_response({"status": "cleared", "request_count": 0})
             case "/":
                 self._send_help_page()
             case _:
@@ -273,10 +350,10 @@ def main() -> None:
     https_port = http_port + 1
 
     # Create HTTP server
-    http_server = HTTPServer(("", http_port), MCPMockHandler)
+    http_server = HTTPServer(("", http_port), MCPMockHandler)  # type: ignore[arg-type]
 
     # Create HTTPS server with self-signed certificate
-    https_server = HTTPServer(("", https_port), MCPMockHandler)
+    https_server = HTTPServer(("", https_port), MCPMockHandler)  # type: ignore[arg-type]
 
     # Generate or load self-signed certificate
     script_dir = Path(__file__).parent

docker-compose-library.yaml

@@ -6,7 +6,7 @@ services:
       dockerfile: dev-tools/mcp-mock-server/Dockerfile
     container_name: mcp-mock-server
     ports:
-      - "3000:3000"
+      - "9000:3000"
     networks:
       - lightspeednet
     healthcheck:
@@ -66,6 +66,11 @@ services:
       - WATSONX_API_KEY=${WATSONX_API_KEY:-}
       # Enable debug logging if needed
       - LLAMA_STACK_LOGGING=${LLAMA_STACK_LOGGING:-}
+    entrypoint: >
+      /bin/bash -c "
+      printf %s 'test-secret-token-123' > /tmp/lightspeed-mcp-test-token &&
+      /app-root/.venv/bin/python3.12 /app-root/src/lightspeed_stack.py
+      "
     healthcheck:
       test: ["CMD", "curl", "-f", "http://localhost:8080/liveness"]
       interval: 10s   # how often to run the check

docker-compose.yaml

@@ -6,7 +6,7 @@ services:
       dockerfile: dev-tools/mcp-mock-server/Dockerfile
     container_name: mcp-mock-server
     ports:
-      - "3000:3000"
+      - "9000:3000"
     networks:
       - lightspeednet
     healthcheck:
@@ -84,6 +84,11 @@ services:
       - TENANT_ID=${TENANT_ID:-}
       - CLIENT_ID=${CLIENT_ID:-}
       - CLIENT_SECRET=${CLIENT_SECRET:-}
+    entrypoint: >
+      /bin/bash -c "
+      printf %s 'test-secret-token-123' > /tmp/lightspeed-mcp-test-token &&
+      /app-root/.venv/bin/python3.12 /app-root/src/lightspeed_stack.py
+      "
     depends_on:
         llama-stack:
           condition: service_healthy

src/app/endpoints/streaming_query.py

@@ -263,6 +263,22 @@ async def retrieve_response_generator(
                 turn_summary,
             )
         # Retrieve response stream (may raise exceptions)
+        # Log request details before calling Llama Stack (MCP debugging)
+        if responses_params.tools is not None and len(responses_params.tools) > 0:
+            # Filter MCP tools once for efficiency
+            mcp_tools = [t for t in responses_params.tools if t.get("type") == "mcp"]
+            if len(mcp_tools) > 0:
+                logger.debug(
+                    "Calling Llama Stack Responses API (streaming) with %d MCP tool(s)",
+                    len(mcp_tools),
+                )
+                # Log MCP server endpoints that may be called
+                logger.debug("MCP server endpoints that may be called:")
+                for tool in mcp_tools:
+                    logger.debug(
+                        "  - %s: %s", tool.get("server_label"), tool.get("server_url")
+                    )
+
         response = await context.client.responses.create(
             **responses_params.model_dump()
         )

src/app/main.py

@@ -20,7 +20,8 @@
 from log import get_logger
 from a2a_storage import A2AStorageFactory
 from models.responses import InternalServerErrorResponse
-from utils.common import register_mcp_servers_async
+
+# from utils.common import register_mcp_servers_async  # Not needed for Responses API
 from utils.llama_stack_version import check_llama_stack_version
 
 logger = get_logger(__name__)
@@ -69,9 +70,32 @@ async def lifespan(_app: FastAPI) -> AsyncIterator[None]:
         )
         raise
 
-    logger.info("Registering MCP servers")
-    await register_mcp_servers_async(logger, configuration.configuration)
-    get_logger("app.endpoints.handlers")
+    # Log MCP server configuration
+    mcp_servers = configuration.configuration.mcp_servers
+    if mcp_servers:
+        logger.info("Loaded %d MCP server(s) from configuration:", len(mcp_servers))
+        for server in mcp_servers:
+            has_auth = bool(server.authorization_headers)
+            logger.info(
+                "  - %s at %s (auth: %s)",
+                server.name,
+                server.url,
+                "yes" if has_auth else "no",
+            )
+            # Debug: Show auth header names if configured
+            if has_auth:
+                logger.debug(
+                    "    Auth headers: %s",
+                    ", ".join(server.authorization_headers.keys()),
+                )
+    else:
+        logger.info("No MCP servers configured")
+
+    # NOTE: MCP server registration not needed for Responses API
+    # The Responses API takes inline tool definitions instead of pre-registered toolgroups
+    # logger.info("Registering MCP servers")
+    # await register_mcp_servers_async(logger, configuration.configuration)
+    # get_logger("app.endpoints.handlers")
     logger.info("App startup complete")
 
     initialize_database()

src/constants.py

@@ -125,6 +125,8 @@
 # MCP authorization header special values
 MCP_AUTH_KUBERNETES = "kubernetes"
 MCP_AUTH_CLIENT = "client"
+# MCP authorization header name (special handling for llama_stack 0.4.x+)
+MCP_AUTHORIZATION_HEADER = "authorization"
 
 # default RAG tool value
 DEFAULT_RAG_TOOL = "knowledge_search"