Building MCP PyExec: Secure Python Execution Server with Docker & Authentication

sequenceDiagram
    participant User as 👤 User/Client
    participant MCP as 🔧 MCP PyExec Server
    participant Auth as 🔐 OAuth IDP
    participant Docker as 🐳 Docker Engine
    participant Container as 📦 Python Container
    participant Session as 💾 Session Storage

    Note over User,Session: User Requests Python Code Execution

    User->>MCP: execute_python(code, session_id?)
    
    Note over MCP: Authentication Check
    MCP->>Auth: Validate JWT Bearer Token
    Auth-->>MCP: ✅ Token Valid (scope: python:execute)
    
    Note over MCP: Session Management
    alt Session ID provided
        MCP->>Session: Check/Create session directory
        Session-->>MCP: Session path ready
    else No session
        MCP->>MCP: Use temporary execution
    end
    
    Note over MCP,Container: Docker Container Setup
    MCP->>Docker: Generate unique container name
    MCP->>Docker: Configure security limits
    Note right of MCP: Memory: 512MB
CPU: 0.5 cores
Network: none
Timeout: 30s
    
    MCP->>Docker: Create container with:
- ipython_wrapper.py
- Volume mount (session)
- Security constraints
    Docker->>Container: Start container
    Container-->>Docker: ✅ Container running
    
    Note over Container: Code Execution
    MCP->>Container: Send Python code via stdin
    Container->>Container: IPython.run_cell(code)
    
    Note over Container: Capture Outputs
    Container->>Container: Capture stdout/stderr
    Container->>Container: Check for matplotlib plots
    Container->>Container: Format results as JSON
    
    alt Execution successful
        Container-->>MCP: {
  "success": true,
  "output": "text output",
  "images": ["base64..."],
  "result": "42"
}
    else Execution error
        Container-->>MCP: {
  "success": false,
  "error": "exception details",
  "traceback": "..."
}
    else Timeout
        Docker->>Container: Kill container (30s limit)
        Container-->>MCP: {"success": false, "error": "timeout"}
    end
    
    Note over MCP,Session: Session Persistence
    alt Session used
        MCP->>Session: Save variables/state to disk
        Session-->>MCP: ✅ State persisted
    end
    
    Note over Docker: Cleanup
    Docker->>Container: Remove container (--rm flag)
    Container-->>Docker: ✅ Container cleaned up
    
    Note over MCP: Response Formatting
    MCP->>MCP: Format MCP tool response
    MCP-->>User: {
  "content": [{
    "type": "text",
    "text": "execution results..."
  }]
}

    Note over User,Session: Complete Execution Cycle

Introduction

This is the story of building MCP PyExec - a Model Context Protocol (MCP) server that allows safe Python code execution. What started as a simple MCP server evolved into a complete ecosystem with three main components:

  1. mcp-pyexec - The core MCP server for Python execution
  2. oauth-idp-server - A custom OAuth Identity Provider for HTTP streaming authentication
  3. mcp-pyexec-client - A testing client to validate the entire system

The Challenge: HTTP Streaming and Authentication

The initial goal was straightforward: create an MCP server that could execute Python code safely. However, the requirement for HTTP streaming support introduced complexity that necessitated building a complete authentication system.

Component 1: MCP PyExec Server

Architecture Overview

The MCP PyExec server is built around a secure, containerized Python execution engine. It consists of three main components:

  1. ipython_server.py - The main FastMCP server that handles MCP protocol requests
  2. ipython_wrapper.py - A lightweight Python script that runs inside Docker containers
  3. Dockerfile - Defines the execution environment with pre-installed data science packages

Core Features

Secure Code Execution

Session Management

The server supports persistent sessions through the session_id parameter:

# Each session gets its own directory
session_dir = os.path.abspath(f'sessions/{session_id}')
# Mounted into containers for state persistence
docker_args.extend(['-v', f"{session_dir}:/home/user/session"])

Sessions allow variables, imports, and state to persist across multiple code executions, making it perfect for interactive data analysis workflows.

Rich Output Support

The system captures multiple types of output:

Technical Implementation

Docker Container Lifecycle

# Generate unique container name
container_name = f"ipython-exec-{uuid.uuid4()}"

# Configure security and resource limits
docker_args = [
    'docker', 'run',
    '--name', container_name,
    '--rm',                    # Auto-cleanup
    '--memory', '512m',        # Memory limit
    '--cpus', '0.5',          # CPU limit
    '--network', 'none',       # No network access
    '-i',                      # Interactive mode
]

IPython Integration

The wrapper script uses IPython’s interactive shell for rich code execution:

def execute_code(code):
    ipython = get_ipython()
    if ipython is None:
        from IPython.core.interactiveshell import InteractiveShell
        ipython = InteractiveShell.instance()
    
    # Capture all outputs
    with redirect_stdout(stdout_capture), redirect_stderr(stderr_capture):
        result = ipython.run_cell(code)

Authentication System

Built-in JWT-based authentication with bearer token support:

auth = BearerAuthProvider(
    jwks_uri="https://idp.objectgraph.com/.well-known/jwks.json",
    algorithm="ES256",
    issuer="https://idp.objectgraph.com"
)

The server supports both development mode (auto-generated keys) and production mode (external JWKS endpoint).

Pre-installed Libraries

The Docker image comes with a comprehensive data science stack:

Error Handling and Safety

The system implements multiple layers of protection:

  1. Process Timeout: Automatic container termination after 30 seconds
  2. Resource Limits: Hard limits on memory and CPU usage
  3. Output Validation: Size checks and encoding safety
  4. Container Cleanup: Automatic removal of completed containers
  5. Non-root Execution: All code runs as a non-privileged user

Real-world Usage Example

# Data analysis with visualization
code = """
import pandas as pd
import matplotlib.pyplot as plt

# Load and analyze data
df = pd.DataFrame({
    'x': range(10),
    'y': [i**2 for i in range(10)]
})

# Create visualization
plt.figure(figsize=(8, 6))
plt.plot(df['x'], df['y'], 'bo-')
plt.title('Quadratic Growth')
plt.show()

print(f"Dataset shape: {df.shape}")
"""

This would return both the text output (“Dataset shape: (10, 2)”) and the generated plot as a base64-encoded PNG image, all properly formatted for MCP protocol consumption.

Component 2: OAuth IDP Server

⚠️ IMPORTANT WARNING: This OAuth IDP server is a development/demonstration tool only. It is NOT intended for production use. For production deployments, use established identity providers like Auth0, Okta, Google, or enterprise solutions. This implementation lacks production-level security hardening, monitoring, and compliance features.

Why Build a Custom IDP?

When implementing HTTP streaming support for the MCP PyExec server, proper authentication became essential. Building a custom OAuth 2.0 Identity Provider was necessary for several reasons:

  1. MCP-Specific Requirements: Need to understand MCP protocol nuances and third-party authorization flows
  2. Development Flexibility: Full control over authentication logic for testing and prototyping
  3. Educational Value: Understanding OAuth 2.0 implementation details firsthand
  4. Third-Party Integration: Supporting delegation to external providers (Google, GitHub, etc.)

Architecture Overview

The OAuth IDP server is a comprehensive RFC-compliant OAuth 2.0 authorization server built with FastAPI:

# Core OAuth 2.0 endpoints
@app.get("/oauth/authorize")     # Authorization endpoint
@app.post("/oauth/token")        # Token exchange  
@app.post("/register")           # Dynamic client registration (RFC 7591)
@app.post("/revoke")             # Token revocation (RFC 7009)
@app.get("/.well-known/oauth-authorization-server")  # Discovery (RFC 8414)

Key Features

RFC-Compliant OAuth 2.0 Implementation

Security Features

Database Schema

The server uses SQLAlchemy with support for SQLite, PostgreSQL, and MySQL:

class User(Base):
    username = Column(String, unique=True)
    email = Column(String, unique=True) 
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)

class OAuthClient(Base):
    client_id = Column(String, unique=True)
    client_secret = Column(String)  # SHA-256 hashed
    redirect_uri = Column(String)
    name = Column(String)

class ThirdPartyProvider(Base):
    name = Column(String, unique=True)
    client_id = Column(String)
    client_secret = Column(String)
    authorization_endpoint = Column(String)
    token_endpoint = Column(String)
    userinfo_endpoint = Column(String)

Third-Party Authorization Flow

One of the most innovative features is support for MCP Third-Party Authorization Flow (2025-03-26 specification):

How It Works

  1. Provider Registration: Admin registers third-party OAuth providers (Google, GitHub, etc.)
  2. Session Creation: Client initiates OAuth with provider parameter
  3. Delegation: User is redirected to third-party provider for authentication
  4. Token Binding: Third-party tokens are securely bound to MCP sessions
  5. MCP Token Generation: Provider-bound MCP tokens are issued

Example Third-Party Flow

# 1. Client initiates OAuth with provider parameter
GET /oauth/authorize?response_type=code&client_id=CLIENT&provider=google

# 2. Server redirects to Google OAuth
GET https://accounts.google.com/o/oauth2/auth?client_id=...

# 3. Google redirects back with code
GET /oauth/third-party/callback/SESSION_ID?code=GOOGLE_CODE

# 4. Server exchanges Google code for tokens and generates MCP token
# 5. Client receives MCP authorization code

JWT Implementation

The server uses ECDSA (ES256) for JWT signing, providing better security than RSA:

# Key generation and management
private_key = ec.generate_private_key(ec.SECP256R1())
public_key = private_key.public_key()

# JWT creation with proper claims
def create_access_token(data: dict, expires_delta: timedelta, scope: str):
    to_encode = data.copy()
    to_encode.update({
        "exp": datetime.utcnow() + expires_delta,
        "iat": datetime.utcnow(),
        "iss": BASE_URL,
        "scope": scope
    })
    headers = {"kid": KEY_ID}
    return jwt.encode(to_encode, private_pem, algorithm="ES256", headers=headers)

JWKS Endpoint

Provides JSON Web Key Set for token validation:

@app.get("/.well-known/jwks.json")
def jwks():
    public_numbers = public_key.public_numbers()
    return {
        "keys": [{
            "kty": "EC",
            "kid": KEY_ID,
            "use": "sig", 
            "alg": "ES256",
            "crv": "P-256",
            "x": int_to_base64url_uint(public_numbers.x),
            "y": int_to_base64url_uint(public_numbers.y)
        }]
    }

Configuration and Deployment

Environment Variables

BASE_URL=https://your-domain.com
SECRET_KEY=your-generated-secret-key
DATABASE_URL=sqlite:///./oauth_idp.db
CORS_ORIGINS=https://your-frontend.com,https://your-api.com
ACCESS_TOKEN_EXPIRE_MINUTES=30

Database Support

Request Logging and Security

The server implements comprehensive request logging with security considerations:

@app.middleware("http")
async def log_requests(request: Request, call_next):
    # Generate unique request ID for tracing
    request_id = str(uuid.uuid4())[:8]
    
    # Redact sensitive fields in logs
    for key in json_body.keys():
        if any(sensitive in key.lower() 
               for sensitive in ["password", "secret", "token"]):
            json_body[key] = "***REDACTED***"

Web Interface

The server includes HTML interfaces for user interaction:

Integration with MCP PyExec

The IDP server specifically supports the MCP PyExec server:

# MCP PyExec server configuration
auth = BearerAuthProvider(
    jwks_uri="https://idp.objectgraph.com/.well-known/jwks.json",
    algorithm="ES256",
    issuer="https://idp.objectgraph.com"
)

Production Considerations

For production use, several enhancements would be needed:

Security Hardening

Monitoring and Compliance

Scalability

Learning Outcomes

Building this OAuth IDP provided deep insights into:

  1. OAuth 2.0 Specification: Understanding RFC requirements and edge cases
  2. JWT Security: ECDSA vs RSA, proper claims handling, key rotation
  3. Third-Party Integration: Provider delegation and token binding patterns
  4. FastAPI Architecture: Middleware, dependency injection, async handling
  5. Database Design: OAuth-specific schema patterns and relationships

The implementation serves as both a functional authentication server for the MCP ecosystem and an educational reference for OAuth 2.0 implementation details.

Component 3: MCP PyExec Client

Purpose and Design Philosophy

The MCP PyExec Client serves as both a testing tool and reference implementation for interacting with the Python execution server. Building a dedicated client was essential for several reasons:

  1. End-to-End Validation: Testing the complete flow from authentication to code execution
  2. OAuth Flow Testing: Validating the custom IDP integration works correctly
  3. HTTP Streaming Verification: Ensuring streaming responses work as expected
  4. Reference Implementation: Showing other developers how to integrate with the server

Core Architecture

The client is built as a lightweight Python application using the FastMCP client library:

from fastmcp import Client
from fastmcp.client.auth import OAuth

async def test_execute_python():
    server_url = "https://idp.objectgraph.com/mcp/"
    
    async with Client(server_url, auth=OAuth(mcp_url=server_url)) as client:
        result = await client.call_tool("execute_python", {"code": python_code})

Authentication Integration

The client demonstrates the complete OAuth flow with the custom IDP:

OAuth Cache Management

The client maintains an OAuth token cache for seamless authentication:

# Cache location: ~/.fastmcp/oauth-mcp-client-cache
# Clean cache when needed: rm -rf ~/.fastmcp/oauth-mcp-client-cache

Automatic Token Handling

Testing Capabilities

Comprehensive Test Suite

The client includes various test scenarios:

# Basic execution test
python_code = """
print("Hello from remote Python execution!")
x = 10
y = 20
result = x + y
print(f"The sum of {x} and {y} is {result}")

# Test library imports
import math
print(f"Square root of 16: {math.sqrt(16)}")

# Return value testing
result
"""

Test Categories

  1. Basic Operations: Simple calculations and print statements
  2. Library Imports: Testing pre-installed package availability
  3. Session Persistence: Multi-call session state validation
  4. Error Handling: Exception and timeout scenarios
  5. Output Types: Text, images, and structured data testing

Usage Examples

Simple Execution Test

# Execute basic Python code
result = await client.call_tool("execute_python", {
    "code": "print('Hello World!')"
})

Data Analysis Workflow

# Test data science capabilities
analysis_code = """
import pandas as pd
import matplotlib.pyplot as plt

# Create sample dataset
data = {'x': range(10), 'y': [i**2 for i in range(10)]}
df = pd.DataFrame(data)

# Generate visualization
plt.plot(df['x'], df['y'])
plt.title('Quadratic Growth')
plt.show()

df.describe()
"""

result = await client.call_tool("execute_python", {"code": analysis_code})

Session Management Testing

# Test 1: Set up variables
await client.call_tool("execute_python", {
    "code": "x = 42\nprint(f'Set x = {x}')",
    "session_id": "test_session"
})

# Test 2: Use previously set variables
await client.call_tool("execute_python", {
    "code": "print(f'Retrieved x = {x}')",
    "session_id": "test_session"
})

Development and Debugging Features

Verbose Output

The client provides detailed logging for debugging:

print("🐍 Executing Python code...")
print(f"Code to execute:\n{python_code}\n")
result = await client.call_tool("execute_python", {"code": python_code})
print("✅ Execution successful!")
print(f"📤 Result: {result}")

Error Handling

Comprehensive error reporting for different failure scenarios:

Configuration Options

Server Endpoints

# Local development
server_url = "http://127.0.0.1:8000/mcp/"

# Production deployment
server_url = "https://idp.objectgraph.com/mcp/"

Authentication Methods

Real-world Testing Scenarios

The client has been used to validate:

  1. Performance Testing: Execute compute-intensive operations
  2. Memory Limits: Test behavior at resource boundaries
  3. Timeout Handling: Validate 30-second execution limits
  4. Concurrent Sessions: Multiple simultaneous session management
  5. Image Generation: Matplotlib plot creation and retrieval
  6. Error Recovery: Exception handling and reporting

Integration Benefits

Having a dedicated client provides several advantages:

The client serves as both a validation tool and a template for other developers building applications that need to integrate with secure Python execution capabilities.

Nginx Configuration

To tie everything together, a proper Nginx configuration was crucial for:

Key Configuration Elements

[Nginx configuration details to be added]

Lessons Learned

Technical Insights

Architectural Decisions

Future Enhancements

Planned Features

Scaling Considerations

Conclusion

Building MCP PyExec was more than just creating a code execution server - it became a comprehensive exploration of the MCP ecosystem, authentication patterns, and distributed system design. The journey highlighted the importance of proper authentication in streaming scenarios and the value of building complete testing infrastructure.

The three-component architecture (server, IDP, client) provides a robust foundation for safe Python code execution within the MCP framework, with lessons applicable to other MCP server implementations.

Code Repositories

This blog post documents the complete journey of building a production-ready MCP server ecosystem from scratch.

By: Gavi Narra on: