First stab - MCP server that uses GCF as a code interpreter

This is just a first stab. It assumes the user has `gcloud` user set up
in the proper project, etc.

It requires some manual setup, use of environment variable for the GCF
URL, etc. But... it works

Author: leblancfg

.github/workflows/test.yml

@@ -0,0 +1,37 @@
+name: Test
+
+on:
+  push:
+    branches: [ main ]
+  pull_request:
+    branches: [ main ]
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.11", "3.12"]
+
+    steps:
+    - uses: actions/checkout@v4
+
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v5
+      with:
+        python-version: ${{ matrix.python-version }}
+
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install -e .
+
+    - name: Run tests
+      run: |
+        pytest -v
+
+    - name: Run linting
+      run: |
+        pip install ruff
+        ruff check src/ tests/
+        ruff format --check src/ tests/

.gitignore

@@ -0,0 +1,10 @@
+# Python-generated files
+__pycache__/
+*.py[oc]
+build/
+dist/
+wheels/
+*.egg-info
+
+# Virtual environments
+.venv

.python-version

@@ -0,0 +1 @@
+3.11

README.md

@@ -0,0 +1,63 @@
+# Code MCP Server
+
+An MCP (Model Context Protocol) server that provides code interpretation capabilities via Google Cloud Functions.
+
+## Features
+
+- Execute Python, JavaScript, and Bash code in a sandboxed environment
+- Automatic deployment to Google Cloud Functions
+- STDIO-based MCP server implementation
+
+## Prerequisites
+
+- Python 3.11+
+- Google Cloud SDK (`gcloud`) installed and configured
+- A Google Cloud Project with Cloud Functions API enabled
+
+## Installation
+
+```bash
+pip install -e ".[dev]"
+```
+
+## Usage
+
+### As an MCP Server
+
+```bash
+python main.py
+```
+
+### Running Tests
+
+```bash
+pytest
+```
+
+### Testing with the MCP Inspector
+You can use the CLI feature with
+
+```console
+$ GCF_URL=$MY_COOL_GCF_URL \
+npx @modelcontextprotocol/inspector@0.11.0 \
+--cli uv run python main.py \
+--method tools/call \
+--tool-name run_code \
+--tool-arg "code=print(1+1)" \
+--tool-arg language=python \
+| jq
+```
+
+## Configuration
+
+Set the `GCF_URL` environment variable to use an existing Cloud Function, otherwise the server will attempt to deploy one automatically.
+
+```bash
+export GCF_URL="https://region-project.cloudfunctions.net/code-interpreter"
+```
+
+## Architecture
+
+- **MCP Server**: Handles tool requests from AI agents
+- **Google Cloud Function**: Executes code in an isolated environment
+- **Supported Languages**: Python, JavaScript (Node.js), Bash

deploy_gcf.py

@@ -0,0 +1,41 @@
+#!/usr/bin/env python
+"""
+Script to deploy the Google Cloud Function for code execution.
+
+Usage: python deploy_gcf.py [--project PROJECT_ID]
+"""
+
+import argparse
+import sys
+from src.code_mcp.gcf_deployer import GCFDeployer
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Deploy Code Interpreter Google Cloud Function")
+    parser.add_argument(
+        "--project",
+        help="Google Cloud Project ID (defaults to current gcloud config)",
+        default=None
+    )
+    
+    args = parser.parse_args()
+    
+    deployer = GCFDeployer(project_id=args.project)
+    
+    print("🚀 Deploying Code Interpreter Cloud Function...")
+    
+    result = deployer.deploy()
+    
+    if result["success"]:
+        print(f"✅ Successfully deployed!")
+        print(f"📍 Function URL: {result['function_url']}")
+        print(f"🔧 Project ID: {result['project_id']}")
+        print(f"\nTo use with the MCP server, set the environment variable:")
+        print(f"export GCF_URL=\"{result['function_url']}\"")
+    else:
+        print(f"❌ Deployment failed: {result.get('error', 'Unknown error')}")
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()

dev.yml

@@ -0,0 +1,11 @@
+name: code-mcp
+
+type: python
+
+up:
+  - uv
+
+commands:
+  setup: python deploy_gcf.py
+  server: python main.py
+  test: pytest . -v

example_usage.py

@@ -0,0 +1,73 @@
+#!/usr/bin/env python
+"""
+Example demonstrating how to use the Code MCP server programmatically.
+
+This shows how an AI agent would interact with the MCP server to execute code.
+"""
+
+import asyncio
+import json
+from mcp.client import ClientSession
+from mcp.client.stdio import stdio_client
+
+
+async def main():
+    # Connect to the MCP server via stdio
+    async with stdio_client() as (read_stream, write_stream):
+        async with ClientSession(read_stream, write_stream) as session:
+            # Initialize the session
+            await session.initialize()
+            
+            # List available tools
+            tools = await session.list_tools()
+            print("Available tools:")
+            for tool in tools:
+                print(f"  - {tool.name}: {tool.description}")
+            
+            # Example 1: Execute Python code
+            print("\n--- Python Example ---")
+            result = await session.call_tool(
+                "run_code",
+                arguments={
+                    "code": "print('Hello from Python!')\nprint(2 + 2)",
+                    "language": "python"
+                }
+            )
+            print(f"Result: {result[0].text}")
+            
+            # Example 2: Execute JavaScript code
+            print("\n--- JavaScript Example ---")
+            result = await session.call_tool(
+                "run_code",
+                arguments={
+                    "code": "console.log('Hello from JavaScript!');\nconsole.log(6 * 7);",
+                    "language": "javascript"
+                }
+            )
+            print(f"Result: {result[0].text}")
+            
+            # Example 3: Execute Bash code
+            print("\n--- Bash Example ---")
+            result = await session.call_tool(
+                "run_code",
+                arguments={
+                    "code": "echo 'Hello from Bash!'\necho $((3 + 4))",
+                    "language": "bash"
+                }
+            )
+            print(f"Result: {result[0].text}")
+            
+            # Example 4: Handle errors
+            print("\n--- Error Handling Example ---")
+            result = await session.call_tool(
+                "run_code",
+                arguments={
+                    "code": "print('Unclosed string",
+                    "language": "python"
+                }
+            )
+            print(f"Result: {result[0].text}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

gcf/__init__.py

@@ -0,0 +1,71 @@
+import subprocess
+import json
+import logging
+from flask import jsonify
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+TIMEOUT = 30
+SUPPORTED_LANGUAGES = {
+    "python": ["python", "-c"],
+    "javascript": ["node", "-e"],
+    "bash": ["bash", "-c"]
+}
+
+
+def execute_code(request):
+    try:
+        request_json = request.get_json()
+        
+        if not request_json:
+            return jsonify({"error": "Invalid request body"}), 400
+        
+        code = request_json.get("code")
+        language = request_json.get("language")
+        
+        if not code:
+            return jsonify({"error": "Missing required field: code"}), 400
+        
+        if not language:
+            return jsonify({"error": "Missing required field: language"}), 400
+        
+        if language not in SUPPORTED_LANGUAGES:
+            return jsonify({"error": f"Unsupported language: {language}"}), 400
+        
+        cmd = SUPPORTED_LANGUAGES[language] + [code]
+        
+        logger.info(f"Executing {language} code")
+        
+        try:
+            result = subprocess.run(
+                cmd,
+                capture_output=True,
+                text=True,
+                timeout=TIMEOUT
+            )
+            
+            return jsonify({
+                "stdout": result.stdout,
+                "stderr": result.stderr,
+                "exitCode": result.returncode
+            }), 200
+            
+        except subprocess.TimeoutExpired:
+            return jsonify({
+                "stdout": "",
+                "stderr": f"Code execution timed out after {TIMEOUT} seconds",
+                "exitCode": -1
+            }), 200
+        
+        except Exception as e:
+            logger.error(f"Execution error: {str(e)}")
+            return jsonify({
+                "stdout": "",
+                "stderr": f"Execution error: {str(e)}",
+                "exitCode": -1
+            }), 200
+    
+    except Exception as e:
+        logger.error(f"Request handling error: {str(e)}")
+        return jsonify({"error": f"Server error: {str(e)}"}), 500

gcf/main.py

@@ -0,0 +1,2 @@
+functions-framework==3.*
+flask==3.*