Skip to content

Latest commit

 

History

History
298 lines (218 loc) · 8.67 KB

File metadata and controls

298 lines (218 loc) · 8.67 KB

MCP Lambda Handler

A Python library for building serverless Model Context Protocol (MCP) HTTP servers on AWS Lambda. Register tools and resources using decorators, plug in session storage, and drop the handler into any Lambda function.

Why this project?

Running an MCP server typically means managing a long-lived process — a container, an EC2 instance, or a persistent service that stays up waiting for connections. That works, but it comes with real operational overhead: you need to provision infrastructure, handle scaling, pay for idle time, and manage availability.

AWS Lambda flips that model. A Lambda function spins up on demand, handles a request, and disappears. You pay only for what you use, AWS handles scaling automatically, and there is no infrastructure to maintain.

The problem is that MCP speaks JSON-RPC over HTTP with session state, and Lambda speaks API Gateway proxy events. Bridging the two correctly — parsing the protocol, routing methods, managing sessions, serialising responses — is tedious boilerplate that every project would otherwise have to write from scratch.

This library does that work for you. You write plain Python functions, decorate them with @mcp.tool(), and point your Lambda handler at mcp.handle_request. The result is a fully compliant MCP HTTP server that:

  • Costs nothing at rest — Lambda charges only per invocation, so an MCP server with light traffic costs fractions of a cent per month
  • Scales to zero and back instantly — no always-on process, no warm-up configuration needed for most workloads
  • Deploys in minutes — zip your code, set the handler, done; no Docker, no ECS, no load balancer to configure
  • Integrates with the AWS ecosystem naturally — your tool functions run inside Lambda, so they have IAM roles, VPC access, environment variables, and direct access to every AWS service
  • Handles sessions without extra infrastructure — pass a DynamoDB table name and per-client session state is managed for you with a 24-hour TTL

Installation

pip install aws-lambda-mcp-server

Quick Start

from awslabs.mcp_lambda_handler import MCPLambdaHandler

mcp = MCPLambdaHandler(name="my-mcp-server", version="1.0.0")

@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b

def lambda_handler(event, context):
    return mcp.handle_request(event, context)

Tools

Type hints become the input schema automatically

from typing import Optional
from awslabs.mcp_lambda_handler import MCPLambdaHandler

mcp = MCPLambdaHandler(name="my-mcp-server")

@mcp.tool()
def search_products(
    query: str,
    max_results: int,
    category: Optional[str],
) -> str:
    """Search the product catalogue.

    Args:
        query: Search keywords
        max_results: Maximum number of results to return
        category: Optional category filter
    """
    # ... your logic here
    return f"Found results for: {query}"

Supported types: str, int, float, bool, List[T], Dict[str, T], Optional[T], and Enum.

Enum parameters

from enum import Enum
from awslabs.mcp_lambda_handler import MCPLambdaHandler

mcp = MCPLambdaHandler(name="my-mcp-server")

class Environment(Enum):
    DEV = "dev"
    STAGING = "staging"
    PROD = "prod"

@mcp.tool()
def deploy(service: str, env: Environment) -> str:
    """Deploy a service to an environment.

    Args:
        service: Name of the service to deploy
        env: Target environment
    """
    return f"Deploying {service} to {env.value}"

Returning images

Return bytes from a tool and the handler automatically base64-encodes them as an image content block. JPEG, PNG, GIF, and WebP are auto-detected.

import boto3
from awslabs.mcp_lambda_handler import MCPLambdaHandler

mcp = MCPLambdaHandler(name="my-mcp-server")

@mcp.tool()
def get_chart(metric: str) -> bytes:
    """Generate a chart for a metric and return it as an image.

    Args:
        metric: The metric name to chart
    """
    s3 = boto3.client("s3")
    obj = s3.get_object(Bucket="my-charts", Key=f"{metric}.png")
    return obj["Body"].read()

Resources

Static resource

from awslabs.mcp_lambda_handler import MCPLambdaHandler
from awslabs.mcp_lambda_handler.types import StaticResource

mcp = MCPLambdaHandler(name="my-mcp-server")

mcp.add_resource(StaticResource(
    uri="config://app/settings",
    name="App Settings",
    content='{"timeout": 30, "retries": 3}',
    description="Application configuration",
    mime_type="application/json",
))

File resource

from awslabs.mcp_lambda_handler import MCPLambdaHandler
from awslabs.mcp_lambda_handler.types import FileResource

mcp = MCPLambdaHandler(name="my-mcp-server")

mcp.add_resource(FileResource(
    uri="docs://openapi",
    path="/var/task/openapi.yaml",
    name="OpenAPI Spec",
    description="API specification",
))

Resource decorator (dynamic content)

import json
import boto3
from awslabs.mcp_lambda_handler import MCPLambdaHandler

mcp = MCPLambdaHandler(name="my-mcp-server")

@mcp.resource(
    uri="data://live-config",
    name="Live Config",
    description="Config fetched from Parameter Store at read time",
    mime_type="application/json",
)
def live_config():
    ssm = boto3.client("ssm")
    value = ssm.get_parameter(Name="/myapp/config")["Parameter"]["Value"]
    return value

Session Management

Stateless (default)

No session store is configured — each request is independent. Session IDs are still issued but nothing is persisted.

from awslabs.mcp_lambda_handler import MCPLambdaHandler

mcp = MCPLambdaHandler(name="my-mcp-server")

DynamoDB sessions

Pass a DynamoDB table name to enable persistent sessions with a 24-hour TTL.

from awslabs.mcp_lambda_handler import MCPLambdaHandler

mcp = MCPLambdaHandler(
    name="my-mcp-server",
    session_store="mcp-sessions",  # DynamoDB table name
)

Or use DynamoDBSessionStore directly:

from awslabs.mcp_lambda_handler import MCPLambdaHandler
from awslabs.mcp_lambda_handler.session import DynamoDBSessionStore

mcp = MCPLambdaHandler(
    name="my-mcp-server",
    session_store=DynamoDBSessionStore(table_name="mcp-sessions"),
)

Reading and writing session data from a tool

from awslabs.mcp_lambda_handler import MCPLambdaHandler

mcp = MCPLambdaHandler(name="my-mcp-server", session_store="mcp-sessions")

@mcp.tool()
def increment_counter() -> str:
    """Increment the per-session counter."""
    session = mcp.get_session()
    count = session.get("count", 0) + 1 if session else 1
    mcp.update_session(lambda s: s.set("count", count))
    return f"Count is now {count}"

@mcp.tool()
def get_counter() -> str:
    """Get the current per-session counter value."""
    session = mcp.get_session()
    count = session.get("count", 0) if session else 0
    return f"Count: {count}"

Custom session backend

Subclass SessionStore to use any storage backend:

from typing import Any, Dict, Optional
from awslabs.mcp_lambda_handler import MCPLambdaHandler
from awslabs.mcp_lambda_handler.session import SessionStore
import uuid

class RedisSessionStore(SessionStore):
    def __init__(self, redis_client):
        self.redis = redis_client

    def create_session(self, session_data: Optional[Dict[str, Any]] = None) -> str:
        session_id = str(uuid.uuid4())
        self.redis.setex(session_id, 86400, json.dumps(session_data or {}))
        return session_id

    def get_session(self, session_id: str) -> Optional[Dict[str, Any]]:
        data = self.redis.get(session_id)
        return json.loads(data) if data else None

    def update_session(self, session_id: str, session_data: Dict[str, Any]) -> bool:
        self.redis.setex(session_id, 86400, json.dumps(session_data))
        return True

    def delete_session(self, session_id: str) -> bool:
        self.redis.delete(session_id)
        return True

mcp = MCPLambdaHandler(
    name="my-mcp-server",
    session_store=RedisSessionStore(redis_client),
)

Example Architecture

A typical deployment looks like:

Client → API Gateway (/mcp)
              │
              ├── Lambda Authorizer  (validates bearer token)
              │
              └── MCP Lambda         (this library)
                        │
                        └── DynamoDB  (optional session store)

The Lambda function receives API Gateway proxy events and returns proxy responses — no extra configuration needed.

Development

git clone https://github.com/awslabs/mcp.git
cd mcp/src/mcp-lambda-handler
pip install -e ".[dev]"
pytest

License

Apache-2.0 — see LICENSE.