mcpbytes-lambda-apigw 0.1.0

AWS API Gateway transport adapter for mcpbytes-lambda-core

mcpbytes-lambda-apigw

Production-ready AWS API Gateway transport adapter for Model Context Protocol (MCP) servers. Seamlessly converts between HTTP requests and MCP JSON-RPC with enterprise-grade authentication and CORS handling.

🚀 Quick Start

from mcpbytes_lambda.core import MCPServer
from mcpbytes_lambda.apigw import ApiGatewayAdapter

# Create your MCP server with tools
mcp = MCPServer(name="my-server", version="1.0.0")

@mcp.tool(name="example.tool")
def my_tool(message: str) -> str:
    return f"Processed: {message}"

# Zero-boilerplate Lambda handler
def lambda_handler(event, context):
    adapter = ApiGatewayAdapter()
    return adapter.handle(event, mcp)  # That's it!

✨ Features

• 🔒 Smart Authentication - Auto-detects Lambda Authorizers, supports custom token validation
• 🌐 CORS Compliance - Full preflight handling with configurable origins
• 📡 MCP Streamable HTTP - Compliant with MCP 2025-06-18 transport specification
• ⚡ High Performance - Optimized for AWS Lambda cold starts and execution
• 🛡️ Input Validation - JSON-RPC 2.0 validation with detailed error responses
• 📊 Production Logging - Structured logging for debugging and monitoring
• 🔄 Base64 Support - Automatic handling of API Gateway base64 encoding

🔐 Authentication

The API Gateway adapter provides flexible authentication with smart auto-detection:

Authentication Flow

1. Lambda Authorizer Detection: Auto-detects and uses existing Lambda Authorizer context
2. Token Format Validation: Validates Bearer token format when auth is required
3. Custom Validation: Executes custom token validation logic (if provided)
4. User Context Access: Provides unified access to authentication context

Usage Patterns

Lambda Authorizer (Auto-detected)

def lambda_handler(event, context):
    adapter = ApiGatewayAdapter(require_auth=True)
    user_info = adapter.get_user_context(event)  # From Lambda Authorizer
    return adapter.handle(event, mcp_server)

Custom Token Validation

import jwt
from typing import Dict, Any, Union

def validate_jwt_token(token: str) -> Union[Dict[str, Any], bool]:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        return {"user_id": payload["sub"], "role": payload["role"]}
    except jwt.InvalidTokenError:
        return False

def lambda_handler(event, context):
    adapter = ApiGatewayAdapter(require_auth=True, token_validator=validate_jwt_token)
    user_info = adapter.get_user_context(event)  # From custom validator
    return adapter.handle(event, mcp_server)

⚙️ Configuration

adapter = ApiGatewayAdapter(
    cors_origin="https://myapp.com",           # CORS origin header
    require_auth=True,                         # Enable authentication
    auth_header="Authorization",               # Auth header name  
    token_validator=my_custom_validator        # Custom validation function
)

Parameters:

• cors_origin: CORS origin header value (default: "*")
• require_auth: Whether to require Bearer token authentication (default: False)
• auth_header: Header name for authentication token (default: "Authorization")
• token_validator: Optional custom token validation function
  • Should return Dict[str, Any] for user context on success
  • Should return False for explicit rejection
  • Should return True for acceptance without context
  • May raise exceptions for validation errors

📋 Requirements

• Python 3.12+
• mcpbytes-lambda-core package
• AWS API Gateway Lambda Proxy Integration
• Bearer token authentication (via Lambda Authorizer or custom validation)

🏗️ Architecture

┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│   API Gateway   │───▶│  APIGW Adapter  │───▶│   MCP Core      │
│   HTTP Request  │    │  (Transport)    │    │   JSON-RPC      │
└─────────────────┘    └─────────────────┘    └─────────────────┘
         │                       │                       │
         │                       ▼                       │
         │              ┌─────────────────┐              │
         │              │  Authentication │              │
         │              │  CORS Handling  │              │
         │              │  Error Mapping  │              │
         │              └─────────────────┘              │
         │                                                │
         ▼                                                ▼
┌─────────────────┐                              ┌─────────────────┐
│   HTTP Response │◀─────────────────────────────│   Tool Results  │
│   (200/500)     │                              │   JSON-RPC      │
└─────────────────┘                              └─────────────────┘

🛠️ Usage Examples

Basic Lambda Handler

def lambda_handler(event, context):
    adapter = ApiGatewayAdapter()
    return adapter.handle(event, mcp)

Lambda Authorizer Integration

def lambda_handler(event, context):
    adapter = ApiGatewayAdapter(require_auth=True)
    
    # Access user context from Lambda Authorizer
    user_info = adapter.get_user_context(event)
    if user_info:
        print(f"User: {user_info.get('principalId')}")
    
    return adapter.handle(event, mcp)

JWT Token Validation

import jwt
import os
from typing import Dict, Any, Union
from mcpbytes_lambda.core.exceptions import AuthenticationError

def validate_jwt(token: str) -> Union[Dict[str, Any], bool]:
    try:
        payload = jwt.decode(token, os.environ["JWT_SECRET"], algorithms=["HS256"])
        return {
            "user_id": payload["sub"],
            "email": payload["email"],
            "role": payload.get("role", "user")
        }
    except jwt.ExpiredSignatureError:
        raise AuthenticationError("Token has expired")
    except jwt.InvalidTokenError:
        return False

def lambda_handler(event, context):
    adapter = ApiGatewayAdapter(
        require_auth=True,
        token_validator=validate_jwt
    )
    
    user_context = adapter.get_user_context(event)
    return adapter.handle(event, mcp)

Multi-Server Routing

# Route to different MCP servers based on path
def lambda_handler(event, context):
    path = event.get("path", "")
    
    if path.startswith("/math"):
        server = math_mcp_server
    elif path.startswith("/weather"):
        server = weather_mcp_server
    else:
        server = default_mcp_server
    
    adapter = ApiGatewayAdapter(require_auth=True)
    return adapter.handle(event, server)

📡 MCP Protocol Compliance

Streamable HTTP Transport (2025-06-18)

✅ HTTP POST for JSON-RPC requests/notifications/responses
✅ HTTP GET for optional Server-Sent Events (SSE) streams
✅ Proper status codes: 200 OK, 202 Accepted, 4xx/5xx errors
✅ MCP-Protocol-Version header support
✅ Bearer token authentication (RFC 6750)
✅ Session management with Mcp-Session-Id
✅ CORS preflight handling
✅ Content negotiation (application/json + text/event-stream)

Lambda Limitations

❌ Server-Sent Events (SSE) streaming not supported
❌ Real-time server-to-client notifications not supported
❌ HTTP GET for stream initiation not supported
❌ Stream resumability not supported

For full streaming support, deploy as a long-running server process.

Required Headers

Client must send:

POST /your-endpoint HTTP/1.1
Content-Type: application/json
Accept: application/json, text/event-stream #BOTH headers
Authorization: Bearer <token> #if needed/enabled
MCP-Protocol-Version: 2025-06-18 / multiple supported

Server responds with:

HTTP/1.1 200 OK
Content-Type: application/json
Access-Control-Allow-Origin: https://example.com

JSON-RPC Format

{
  "jsonrpc": "2.0",
  "method": "tools/call",
  "params": {
    "name": "example.tool",
    "arguments": {"message": "Hello MCP!"}
  },
  "id": "request-1"
}

🔍 Error Handling

HTTP Status Codes

• 200 OK - CORS preflight successful, JSON-RPC requests processed
• 400 Bad Request - Malformed requests, unsupported protocol versions
• 401 Unauthorized - Missing or invalid Bearer token
• 403 Forbidden - Origin not allowed (CORS violation)
• 406 Not Acceptable - Accept header doesn't include required content types
• 415 Unsupported Media Type - Content-Type is not application/json
• 500 Internal Server Error - JSON-RPC error or server failure

JSON-RPC Error Codes

# Standard JSON-RPC 2.0 error codes
-32700  # Parse error (invalid JSON)
-32600  # Invalid request (missing method, bad headers)
-32601  # Method not found  
-32602  # Invalid params
-32603  # Internal error

🛡️ Security Best Practices

Authentication Strategies

1. Lambda Authorizer (Recommended for production)

   • Centralized authentication logic
   • Automatic caching and performance optimization
   • Built-in AWS integration

2. Custom Token Validation

   • Flexible validation logic
   • Support for JWT, API keys, custom tokens
   • Direct integration with your auth system

Environment Variables

# Production CORS configuration
ALLOWED_ORIGINS="https://example.com,https://app.example.com"

# JWT Secret for token validation
JWT_SECRET="your-secret-key"

# Development (allows all origins - not recommended for production)
# ALLOWED_ORIGINS=""  # Empty = allow all

API Gateway Setup

# SAM Template example
Resources:
  McpApi:
    Type: AWS::Serverless::Api
    Properties:
      Cors:
        AllowMethods: "'POST,OPTIONS'"
        AllowHeaders: "'Content-Type,Authorization,MCP-Protocol-Version'"
        AllowOrigin: "'*'"  # Override with adapter's origin validation
      Auth:
        DefaultAuthorizer: TokenAuth
        Authorizers:
          TokenAuth:
            FunctionArn: !GetAtt AuthFunction.Arn

🚀 Deployment

SAM Template

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31

Resources:
  McpFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: .
      Handler: handler.lambda_handler
      Runtime: python3.12
      Environment:
        Variables:
          ALLOWED_ORIGINS: "https://myapp.com,https://admin.myapp.com"
          JWT_SECRET: !Ref JWTSecret
      Events:
        McpApi:
          Type: Api
          Properties:
            Path: /mcp
            Method: any

CDK Deployment

from aws_cdk import aws_lambda as lambda_
from aws_cdk import aws_apigateway as apigw

lambda_function = lambda_.Function(
    self, "McpFunction",
    runtime=lambda_.Runtime.PYTHON_3_12,
    handler="handler.lambda_handler",
    code=lambda_.Code.from_directory("src/"),
    environment={
        "ALLOWED_ORIGINS": "https://myapp.com",
        "JWT_SECRET": jwt_secret.secret_value_from_json("secret").to_string()
    }
)

api = apigw.RestApi(self, "McpApi")
api.root.add_proxy(default_integration=apigw.LambdaIntegration(lambda_function))

🤝 Contributing

1. Fork the repository
2. Create a feature branch: git checkout -b feature-name
3. Make your changes and add tests
4. Run tests: python -m pytest
5. Submit a pull request

📄 License

Apache 2.0 License - see LICENSE for details.

🔗 Related Packages

• mcpbytes-lambda-core - Transport-agnostic MCP server core
• mcpbytes-lambda-stdio - Stdio transport adapter

📚 Documentation

• MCP Specification
• AWS API Gateway Documentation
• AWS Lambda Proxy Integration
• Project Examples

---

Built with ❤️ for the MCP ecosystem