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

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
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

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 authentication flows
  2. Development Flexibility: Full control over authentication logic for testing and prototyping
  3. Educational Value: Understanding OAuth 2.0 implementation details firsthand
  4. JWT Integration: Seamless integration with MCP server bearer token authentication

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("/authorize")           # Authorization endpoint
@app.post("/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

Core 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 AuthorizationCode(Base):
    code = Column(String, unique=True)
    client_id = Column(String)
    user_id = Column(Integer)
    redirect_uri = Column(String)
    expires_at = Column(DateTime)
    used = Column(Boolean, default=False)

Standard OAuth 2.0 Flow

The implementation follows the standard OAuth 2.0 authorization code flow:

How It Works

  1. Client Registration: OAuth clients register with the IDP server
  2. Authorization Request: Client redirects user to authorization endpoint
  3. User Authentication: User provides username/password credentials
  4. Authorization Code: Server generates and returns authorization code
  5. Token Exchange: Client exchanges code for access token

Example OAuth Flow

# 1. Client initiates OAuth authorization
GET /authorize?response_type=code&client_id=CLIENT_ID&redirect_uri=CALLBACK

# 2. User authenticates with username/password
POST /oauth/login
{username: "user", password: "pass", client_id: "CLIENT_ID"}

# 3. Server redirects with authorization code
GET CALLBACK?code=AUTH_CODE&state=STATE

# 4. Client exchanges code for access token
POST /token
{grant_type: "authorization_code", code: "AUTH_CODE", client_id: "CLIENT_ID"}

# 5. Server returns JWT access token
{access_token: "JWT_TOKEN", token_type: "bearer", expires_in: 1800}

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

MCP Authorization Integration

Our OAuth IDP implementation integrates with the MCP Authorization Specification (2025-03-26), providing the authentication foundation for MCP servers.

OAuth Flow with MCP Server

sequenceDiagram
    participant Client as MCP Client
    participant IDP as OAuth IDP Server
    participant MCP as MCP PyExec Server
    
    Note over Client,MCP: MCP Authentication Flow
    
    Client->>IDP: GET /authorize?client_id=...
    IDP->>IDP: Show login form
    IDP->>IDP: User authenticates (username/password)
    IDP->>Client: Redirect with authorization code
    Client->>IDP: POST /token (exchange code)
    IDP-->>Client: JWT access_token
    
    Note over Client,MCP: Authenticated API Requests
    Client->>MCP: execute_python(code) + Bearer token
    MCP->>IDP: Validate token via JWKS endpoint
    IDP-->>MCP: Token valid + user info
    MCP->>MCP: Execute Python code
    MCP-->>Client: Execution results

Key Implementation Points

1. OAuth 2.0 Core Features

# Our tested implementation includes:
- Dynamic client registration (RFC 7591)
- Authorization code flow (RFC 6749)
- Token revocation (RFC 7009) 
- Authorization server metadata (RFC 8414)
- ECDSA JWT signing (ES256)
- JWKS endpoint for token validation

2. Security Requirements Met

OAuth RequirementOur ImplementationStatus
HTTPS endpoints✅ Production deployment on HTTPS✅ Tested
Token expiration✅ Configurable token lifetime✅ Tested
Secure token storage✅ Database-backed with hashed secrets✅ Tested
Redirect URI validation✅ Base URI matching with security checks✅ Tested
JWT signature validation✅ ECDSA ES256 with JWKS endpoint✅ Tested

3. Discovery Endpoint (RFC 8414)

@app.get("/.well-known/oauth-authorization-server")
def oauth_metadata():
    return {
        "issuer": BASE_URL,
        "authorization_endpoint": f"{BASE_URL}/authorize",
        "token_endpoint": f"{BASE_URL}/token",
        "jwks_uri": f"{BASE_URL}/.well-known/jwks.json",
        "scopes_supported": ["profile", "email"],
        "response_types_supported": ["code"]
    }

Learning Outcomes

Building this OAuth IDP provided deep insights into:

  1. OAuth 2.0 Specification: Understanding RFC requirements and implementation details
  2. JWT Security: ECDSA vs RSA, proper claims handling, key rotation
  3. MCP Integration: How authentication fits into the MCP ecosystem
  4. FastAPI Architecture: Middleware, dependency injection, async handling
  5. Database Design: OAuth-specific schema patterns and relationships
  6. Security Best Practices: Token validation, redirect URI security, password hashing

The implementation serves as both a functional authentication server for the MCP PyExec project and an educational reference for OAuth 2.0 implementation with MCP integration.

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

The idp.objectgraph.com nginx file

server {
    server_name idp.objectgraph.com; 
    
    root /home/ubuntu/oauth-idp-server/static;  
    location = /favicon.ico { access_log off; log_not_found off; }
    location = /mcp {
	rewrite ^/mcp$ /mcp/ last;
    }
 
    # Route /mcp to pyexec service
    location ~ ^/(sse|messages|mcp)(/.*)?$  {
        proxy_set_header Host $http_host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
        proxy_pass http://127.0.0.1:8000;
    }
    
    # Default route for everything else
    location / {
        proxy_set_header Host $http_host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
        proxy_pass http://unix:/home/ubuntu/oauth-idp-server/idp.sock;
    }

    listen 80;
}

look at the key location = /mcp, this is important to ensure /mcp POST is not converted as as HTTP 307 /mcp/ GET request

The pyexec.objectgraph.com nginx is below

server {
    server_name pyexec.objectgraph.com;

    root /home/ubuntu/mcp-pyexec/static;  # Create this directory if you have static files
    location = /favicon.ico { access_log off; log_not_found off; }

    location / {
        proxy_set_header Host $http_host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
        proxy_pass http://127.0.0.1:8000;
    }
    listen 80;
}

Both these sites are fronted with cloudflare

Lessons Learned

Technical Insights

Architectural Decisions

Future Enhancements

Planned Features

Scaling Considerations

Resources

Validating your generated jwt with jwks

https://jwt.davetonge.co.uk/

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

By: Gavi Narra on: