Firewall MCP in Cloud Run Single

cloud-run-single/1-apps/README.md

@@ -24,6 +24,7 @@ Once the applications have been deployed, learn sample commands to test them:
 - [Chat](./apps/chat/README.md)
 - [ADK](./apps/adk/README.md)
 - [Gemma3](./apps/gemma/README.md)
+- [Firewall MCP](./apps/firewall-mcp/README.md)
 
 ## I have not used 0-projects
 

cloud-run-single/1-apps/apps/firewall-mcp/Dockerfile

@@ -0,0 +1,42 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+FROM python:3.12-slim
+
+# Copy uv from the official image
+COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/
+
+# Set the working directory in the container
+WORKDIR /app
+
+# Enable bytecode compilation
+ENV UV_COMPILE_BYTECODE=1
+
+# Allow statements and log messages to immediately appear in the Knative logs
+ENV PYTHONUNBUFFERED True
+
+# Installing separately from its dependencies allows optimal layer caching
+COPY . /app
+#RUN uv sync --locked
+RUN uv sync
+
+# Place executables in the environment at the front of the path
+ENV PATH="/app/.venv/bin:$PATH"
+
+# Expose the port the app runs on (Cloud Run uses the PORT env var)
+# Default is 8080
+EXPOSE 8080
+
+# Cloud Run injects the PORT environment variable.
+CMD ["uv", "run", "server.py"]

cloud-run-single/1-apps/apps/firewall-mcp/README.md

@@ -0,0 +1,15 @@
+# Firewall MCP Server for Cloud Run
+
+A Model Context Protocol (MCP) server for managing Google Cloud Firewall rules, built with FastMCP.
+
+## Use the application
+
+You can test the server using the included client script:
+
+```shell
+# List available tools
+uv run client.py https://YOUR_DOMAIN/mcp list
+
+# Call a tool
+uv run client.py https://YOUR_DOMAIN/mcp call list_firewall_rules --arg project_id=YOUR_PROJECT_ID
+```

cloud-run-single/1-apps/apps/firewall-mcp/client.py

@@ -0,0 +1,161 @@
+import asyncio
+import argparse
+import sys
+import logging
+import json
+from typing import Optional, Dict, Any
+from urllib.parse import urlparse
+
+import subprocess
+from fastmcp import Client
+from fastmcp.client.transports import StreamableHttpTransport
+from fastmcp.client.auth import BearerAuth
+
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(levelname)s - %(message)s'
+)
+logger = logging.getLogger(__name__)
+
+def get_id_token(url: str) -> Optional[str]:
+    """
+    Get a Google Cloud ID Token by invoking the gcloud CLI.
+    This avoids direct dependencies on google-auth libraries.
+    """
+    try:
+        result = subprocess.run(
+            ["gcloud", "auth", "print-identity-token"],
+            capture_output=True,
+            text=True,
+            check=True
+        )
+        token = result.stdout.strip()
+        return token
+
+    except subprocess.CalledProcessError as e:
+        logger.error(f"Error fetching ID token via gcloud: {e.stderr}")
+        return None
+    except FileNotFoundError:
+        logger.error("gcloud CLI not found. Please ensure Google Cloud SDK is installed and in your PATH.")
+        return None
+
+def get_access_token() -> Optional[str]:
+    """Get a Google Cloud Access Token (OAuth2) for API calls."""
+    try:
+        result = subprocess.run(
+            ["gcloud", "auth", "print-access-token"],
+            capture_output=True,
+            text=True,
+            check=True
+        )
+        return result.stdout.strip()
+    except Exception as e:
+        logger.error(f"Error fetching Access token: {e}")
+        return None
+
+async def run_client(url: str, prompt: Optional[str] = None, tool_name: Optional[str] = None, tool_args: Optional[Dict[str, Any]] = None):
+    """
+    Connect to the MCP server via HTTP (Streamable) and interact using FastMCP Client.
+    """
+
+    logger.info(f"Connecting to {url}...")
+
+    auth = None
+    headers = {}
+
+    if url.startswith("http"):
+        logger.info("Fetching ID token for Cloud Run authentication...")
+        id_token = get_id_token(url)
+        if id_token:
+            auth = BearerAuth(id_token)
+            logger.info("Successfully retrieved ID token.")
+        else:
+            logger.warning("Continuing without ID Token. Cloud Run might reject the request.")
+
+        logger.info("Fetching Access token for Google Cloud API calls...")
+        access_token = get_access_token()
+        if access_token:
+            headers["X-GCP-Access-Token"] = access_token
+            logger.info("Successfully retrieved Access token.")
+        else:
+            logger.warning("Continuing without Access Token. Backend API calls might fail.")
+
+    try:
+        transport = StreamableHttpTransport(url=url, auth=auth, headers=headers)
+        async with Client(transport) as client:
+            logger.info("Connected to MCP Server!")
+
+            # List tools to verify connection
+            logger.info("Listing available tools...")
+            tools = await client.list_tools()
+            for t in tools:
+                print(f"- {t.name}: {t.description}")
+
+            # Call tool if requested
+            if tool_name:
+                logger.info(f"Calling tool: {tool_name} with args {tool_args}")
+                result = await client.call_tool(tool_name, arguments=tool_args or {})
+                print("\n--- Result ---")
+                if hasattr(result, "content"):
+                     for content in result.content:
+                        if hasattr(content, "text"):
+                            print(content.text)
+                        else:
+                            print(content)
+                else:
+                    print(result)
+
+    except Exception as e:
+        logger.error(f"Connection failed: {e}")
+        if hasattr(e, "response") and e.response:
+             try:
+                 print(f"\n[Debug] Response status: {e.response.status_code}")
+                 print(f"[Debug] Response headers: {e.response.headers}")
+                 print(f"[Debug] Response text: {e.response.text}")
+             except Exception:
+                 pass
+
+        if "401" in str(e) or "403" in str(e):
+             print("\n[!] Authentication failed. Try running: `gcloud auth print-identity-token` to verify you can get a token.")
+
+def main():
+    parser = argparse.ArgumentParser(description="Authenticated MCP Test Client (FastMCP HTTP)")
+    parser.add_argument("url", help="URL of the MCP server (e.g., https://service.run.app/mcp)")
+
+    subparsers = parser.add_subparsers(dest="command", help="Command to run")
+    subparsers.add_parser("list", help="List available tools")
+
+    call_parser = subparsers.add_parser("call", help="Call a specific tool")
+    call_parser.add_argument("tool_name", help="Name of the tool to call")
+    call_parser.add_argument("--args", help="JSON string of arguments", default="{}")
+    call_parser.add_argument("--arg", action="append", help="Key=Value argument (can be used multiple times)")
+
+    args = parser.parse_args()
+
+    if not args.command:
+        args.command = "list"
+
+    tool_args = {}
+    if args.command == "call":
+        if args.args:
+            try:
+                tool_args = json.loads(args.args)
+            except json.JSONDecodeError:
+                logger.error("Invalid JSON in --args")
+                sys.exit(1)
+
+        if args.arg:
+            for item in args.arg:
+                if "=" in item:
+                    k, v = item.split("=", 1)
+                    tool_args[k] = v
+
+    asyncio.run(run_client(
+        args.url, 
+        tool_name=args.tool_name if args.command == "call" else None,
+        tool_args=tool_args
+    ))
+
+if __name__ == "__main__":
+    main()

cloud-run-single/1-apps/apps/firewall-mcp/pyproject.toml

@@ -0,0 +1,15 @@
+[project]
+name = "firewall-mcp"
+version = "0.1.0"
+description = "MCP Server for Google Cloud Firewall config"
+authors = [{ name = "Google LLC" }]
+license = { text = "Apache-2.0" }
+readme = "README.md"
+requires-python = ">=3.12"
+dependencies = [
+    "fastmcp>=2.0.0",
+    "google-cloud-compute>=1.0.0",
+]
+
+[dependency-groups]
+dev = []