MCP Deployment AWS

A guide and example code for deploying MCP servers cost-effectively on AWS and integrating them with AI agent frameworks.

Cost-Efficient MCP Server Deployment on AWS for AI Agents

A comprehensive guide to deploying Model Context Protocol (MCP) servers on AWS services cost-effectively and integrating them with AI agent frameworks like Google ADK.

Model Context Protocol (MCP)is like a universal translator that allows AI applications (like Claude, ChatGPT, or Google's AI agents) to connect to external tools and data sources in a standardized way.

- Without MCP: Each AI tool needs custom code to connect to databases, APIs, or services
- With MCP: AI tools use a standard "language" to talk to any MCP server, regardless of what it connects to

Instead of building custom integrations for every tool, you create one MCP server that can:

- Query your company's database
- Call external APIs
- Analyze files
- Perform calculations

Then any MCP-compatible AI agent can use all these capabilities instantly.

- Plug-and-Play: Connect any AI agent to any tool through a standard interface
- Real-Time Communication: Live data exchange with Server-Sent Events (SSE)
- Secure and Auditable: Built-in access control and comprehensive logging
- Highly Extensible: Easy to add new capabilities without changing AI agent code
- Cost-Effective: Share one MCP server across multiple AI agents and applications

- Basic understanding of AWS services (Lambda, EC2, or containers)
- Python programming experience
- Basic command line usage
- Understanding of APIs and JSON

- Active AWS account with billing enabled
- AWS CLI installed and configured
- Basic IAM permissions for Lambda/Fargate deployment

- Python 3.9+ installed
- Node.js 18+ (for CDK examples)
- Docker (for containerized deployments)

- Quick Start: 30 minutes
- Full Implementation: 2-4 hours
- Production Deployment: 1-2 days

Step 1: Choose Your Deployment Method (5 minutes)

For Beginners: Start with AWS Lambda (pay-per-use, no servers to manage)For Production: Consider AWS Fargate (always-on, more predictable costs)

Do you expect < 1 million requests per month? ├─ YES → Use AWS Lambda ($10-50/month) └─ NO → Do you need 24/7 availability? ├─ YES → Use AWS Fargate ($50-200/month) └─ NO → Use AWS Lambda with provisioned concurrency

Step 3: Your First MCP Server (15 minutes)

We'll create a simple MCP server that can analyze text and get current weather data.

Test locally, then deploy to AWS with a single command.

flowchart LR subgraph "Your Computer" Host["Host with MCP Client\n(Claude, IDEs, Tools)"] S1["MCP Server A"] S2["MCP Server B"] S3["MCP Server C"] Host <-->|"MCP Protocol"| S1 Host <-->|"MCP Protocol"| S2 Host <-->|"MCP Protocol"| S3 S1 <--> D1[("Local\nData Source A")] S2 <--> D2[("Local\nData Source B")] end subgraph "Internet" S3 <-->|"Web APIs"| D3[("Remote\nService C")] end

- MCP Host: The LLM-powered application (Claude, Google ADK agents)
- MCP Client: Maintains 1:1 connection with MCP Server
- MCP Server: Supplies context, tools, and prompts to the client

- Pay only for execution time (no idle costs)
- Automatic scaling
- No server management
- Ideal for variable/unpredictable workloads
- Cold start optimizations available

- 15-minute execution limit
- Cold start latency (100ms-5 seconds)
- Limited to 10GB memory
- Complex for long-running processes

- Event-driven MCP servers
- Sporadic usage patterns
- Cost-sensitive deployments
- Quick prototyping

- No execution time limits
- Better for long-running processes
- More control over runtime environment
- No cold starts once running
- Up to 120GB memory, 16 vCPU

- Higher costs due to continuous resource allocation
- Longer startup times (35 seconds - 2 minutes)
- More complex setup
- Pay for allocated resources even if idle

- Enterprise deployments requiring high availability
- Long-running MCP servers
- Complex security requirements
- Consistent workloads

- Maximum control and customization
- Cost-effective for consistent high usage
- Can use Reserved Instances for savings

- Requires server management
- Higher operational overhead
- Not truly serverless

- High-volume, consistent workloads
- Custom infrastructure requirements

Cost = (Number of Requests × $0.20 per 1M requests) + (Duration × Memory × $0.0001667 per GB-second)

Example: 1M requests/month, 1GB memory, 2-second average duration

- Requests: 1M × $0.20/1M = $0.20
- Compute: 1M × 2s × 1GB × $0.0001667 = $333.40
- Total: ~$333.60/month

Cost = (vCPU hours × $0.04048) + (GB memory hours × $0.004445)

Example: 1 vCPU, 2GB memory, running 24/7

- vCPU: 744 hours × $0.04048 = $30.12
- Memory: 744 hours × 2GB × $0.004445 = $6.61
- Total: ~$36.73/month

🟢 USE LAMBDA WHEN: ✓ You're starting out or prototyping ✓ Traffic is unpredictable or sporadic ✓ You want minimal operational overhead ✓ Cost optimization is priority #1 ✓ You can handle 100ms-5s cold starts 🟡 USE FARGATE WHEN: ✓ You need consistent sub-100ms response times ✓ Traffic is predictable and consistent ✓ You're running 24/7 workloads ✓ You need complex networking or security ✓ You have containerized applications

Phase 1: Create Your First MCP Server (Local Development)

mkdir my-mcp-server cd my-mcp-server # Create virtual environment python -m venv venv source venv/bin/activate # On Windows: venv\Scripts\activate # Install dependencies pip install mcp anthropic-mcp-server boto3

2. Create a simple MCP server (mcp_server.py)

#!/usr/bin/env python3 """ Simple MCP Server for AWS - Beginner Example This server provides basic text analysis and AWS cost checking tools """ import asyncio import logging from mcp.server import Server from mcp.server.stdio import stdio_server import mcp.types as types # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Create the MCP server app = Server("my-first-mcp-server") @app.list_tools() async def list_tools() -> list[types.Tool]: """List available tools for the MCP client""" return [ types.Tool( name="text-analyzer", description="Analyze text for word count, sentiment, and key phrases", inputSchema={ "type": "object", "properties": { "text": { "type": "string", "description": "Text to analyze" }, "analysis_type": { "type": "string", "enum": ["word_count", "sentiment", "summary"], "description": "Type of analysis to perform" } }, "required": ["text", "analysis_type"] } ), types.Tool( name="aws-simple-cost-check", description="Get basic AWS cost information (demo version)", inputSchema={ "type": "object", "properties": { "service": { "type": "string", "description": "AWS service to check (e.g., 'lambda', 'ec2')" } }, "required": ["service"] } ) ] @app.call_tool() async def call_tool(name: str, arguments: dict) -> list[types.TextContent]: """Handle tool calls from the MCP client""" if name == "text-analyzer": text = arguments["text"] analysis_type = arguments["analysis_type"] if analysis_type == "word_count": word_count = len(text.split()) char_count = len(text) result = f"Word count: {word_count}\nCharacter count: {char_count}" elif analysis_type == "sentiment": # Simple sentiment analysis (in production, use a proper NLP library) positive_words = ["good", "great", "excellent", "amazing", "wonderful"] negative_words = ["bad", "terrible", "awful", "horrible", "disappointing"] text_lower = text.lower() positive_count = sum(1 for word in positive_words if word in text_lower) negative_count = sum(1 for word in negative_words if word in text_lower) if positive_count > negative_count: sentiment = "Positive" elif negative_count > positive_count: sentiment = "Negative" else: sentiment = "Neutral" result = f"Sentiment: {sentiment}\nPositive indicators: {positive_count}\nNegative indicators: {negative_count}" elif analysis_type == "summary": sentences = text.split('. ') result = f"Text summary:\n- Total sentences: {len(sentences)}\n- First sentence: {sentences[0] if sentences else 'No sentences found'}" return [types.TextContent(type="text", text=result)] elif name == "aws-simple-cost-check": service = arguments["service"] # Mock cost data (in production, use boto3 and AWS Cost Explorer) mock_costs = { "lambda": "$12.50 this month (estimated)", "ec2": "$45.30 this month (estimated)", "s3": "$8.75 this month (estimated)" } cost_info = mock_costs.get(service.lower(), "Cost data not available for this service") result = f"AWS {service.upper()} costs: {cost_info}" return [types.TextContent(type="text", text=result)] raise ValueError(f"Unknown tool: {name}") async def main(): """Run the MCP server""" logger.info("Starting MCP server...") async with stdio_server() as streams: await app.run( streams[0], streams[1], app.create_initialization_options() ) if __name__ == "__main__": asyncio.run(main())

# Test the server python mcp_server.py # In another terminal, you can test with: echo '{"jsonrpc": "2.0", "id": 1, "method": "tools/list"}' | python mcp_server.py

Phase 2: Deploy to AWS Lambda (Production Ready)

# Create requirements.txt cat > requirements.txt << EOF mcp>=1.0.0 boto3>=1.26.0 anthropic-mcp-server>=0.1.0 EOF # Create lambda deployment package mkdir lambda_package pip install -r requirements.txt -t lambda_package/ cp mcp_server.py lambda_package/

2. Create Lambda-compatible handler (lambda_package/lambda_handler.py)

import json import asyncio import logging from mcp_server import app logger = logging.getLogger() logger.setLevel(logging.INFO) def lambda_handler(event, context): """AWS Lambda handler for MCP server""" try: # Extract MCP request from Lambda event if 'body' in event: # API Gateway event mcp_request = json.loads(event['body']) else: # Direct Lambda invocation mcp_request = event # Process MCP request response = asyncio.run(process_mcp_request(mcp_request)) return { 'statusCode': 200, 'headers': { 'Content-Type': 'application/json', 'Access-Control-Allow-Origin': '' }, 'body': json.dumps(response) } except Exception as e: logger.error(f"Error processing MCP request: {str(e)}") return { 'statusCode': 500, 'body': json.dumps({'error': str(e)}) } async def process_mcp_request(request): """Process MCP request and return response""" method = request.get('method') params = request.get('params', {}) request_id = request.get('id') try: if method == 'tools/list': tools = await app.list_tools() result = [tool.model_dump() for tool in tools] elif method == 'tools/call': tool_name = params.get('name') arguments = params.get('arguments', {}) result = await app.call_tool(tool_name, arguments) result = [content.model_dump() for content in result] else: raise ValueError(f"Unknown method: {method}") return { 'jsonrpc': '2.0', 'id': request_id, 'result': result } except Exception as e: return { 'jsonrpc': '2.0', 'id': request_id, 'error': { 'code': -32603, 'message': str(e) } }

# Create deployment zip cd lambda_package zip -r ../mcp-server-lambda.zip . cd .. # Create Lambda function aws lambda create-function \ --function-name my-mcp-server \ --runtime python3.11 \ --role arn:aws:iam::YOUR-ACCOUNT:role/lambda-execution-role \ --handler lambda_handler.lambda_handler \ --zip-file fileb://mcp-server-lambda.zip \ --timeout 300 \ --memory-size 512

{ "mcpServers": { "my-mcp-server": { "command": "python", "args": ["/path/to/your/mcp_server.py"] } } }

import requests def call_mcp_server(tool_name, arguments): """Call MCP server via HTTP API""" payload = { "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": tool_name, "arguments": arguments } } response = requests.post( "YOUR-LAMBDA-API-URL", json=payload, headers={"Content-Type": "application/json"} ) return response.json() # Example usage result = call_mcp_server("text-analyzer", { "text": "This is a great example of MCP!", "analysis_type": "sentiment" }) print(result)

# Basic MCP Server on Lambda import json import asyncio from mcp.server import Server from mcp.server.stdio import stdio_server import mcp.types as types app = Server("aws-lambda-mcp-server") @app.list_tools() async def list_tools() -> list[types.Tool]: return [ types.Tool( name="aws-cost-analyzer", description="Analyze AWS costs and usage", inputSchema={ "type": "object", "properties": { "service": {"type": "string"}, "region": {"type": "string"} } } ) ] @app.call_tool() async def call_tool(name: str, arguments: dict) -> list[types.TextContent]: if name == "aws-cost-analyzer": # Implement cost analysis logic result = analyze_aws_costs(arguments) return [types.TextContent(type="text", text=result)] raise ValueError(f"Unknown tool: {name}") def lambda_handler(event, context): # Lambda handler implementation return asyncio.run(handle_mcp_request(event))

# docker-compose.yml for Fargate deployment version: '3.8' services: mcp-server: build: . ports: - "8080:8080" environment: - AWS_REGION=us-east-1 - LOG_LEVEL=INFO healthcheck: test: ["CMD", "curl", "-f", "http://localhost:8080/health"] interval: 30s timeout: 10s retries: 3

# Make sure you're in the virtual environment source venv/bin/activate # Install the correct package pip install mcp>=1.0.0 # Alternative: Try the anthropic package pip install anthropic-mcp-server

# Create basic Lambda execution role aws iam create-role \ --role-name lambda-execution-role \ --assume-role-policy-document '{ "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": {"Service": "lambda.amazonaws.com"}, "Action": "sts:AssumeRole" } ] }' # Attach basic execution policy aws iam attach-role-policy \ --role-name lambda-execution-role \ --policy-arn arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole

# Required methods for basic MCP server @app.list_tools() # For tools/list @app.call_tool() # For tools/call @app.list_resources() # Optional: for resources/list @app.list_prompts() # Optional: for prompts/list

# Example: Add S3 access aws iam attach-role-policy \ --role-name lambda-execution-role \ --policy-arn arn:aws:iam::aws:policy/AmazonS3ReadOnlyAccess

# Test MCP server locally python mcp_server.py # Test with curl (for HTTP endpoints) curl -X POST your-api-endpoint \ -H "Content-Type: application/json" \ -d '{"jsonrpc":"2.0","id":1,"method":"tools/list"}' # Check CloudWatch logs for Lambda aws logs describe-log-groups --log-group-name-prefix /aws/lambda/my-mcp-server

# Initialize expensive resources outside the handler import boto3 # Global initialization (runs once per container) s3_client = boto3.client('s3') dynamodb = boto3.resource('dynamodb') def lambda_handler(event, context): # Handler runs for each request # Use pre-initialized clients pass

import json from functools import lru_cache @lru_cache(maxsize=100) def expensive_operation(input_data): """Cache expensive operations""" # Your expensive logic here return result

import boto3 from botocore.config import Config # Configure connection pooling config = Config( max_pool_connections=10, retries={'max_attempts': 3} ) client = boto3.client('dynamodb', config=config)

flowchart TD A[Google ADK Agent] --> B[MCP Client] B --> C[AWS Lambda MCP Server] B --> D[AWS Fargate MCP Server] C --> E[AWS Services] D --> F[External APIs] E --> G[DynamoDB] E --> H[S3] E --> I[CloudWatch]

# Google ADK agent with MCP integration from google.adk import Agent, Tool from mcp_client import MCPClient class AWSMCPTool(Tool): def __init__(self): self.mcp_client = MCPClient() super().__init__( name="aws-mcp-tool", description="Access AWS services via MCP server" ) async def execute(self, kwargs): # Connect to MCP server on AWS await self.mcp_client.connect("https://your-mcp-server.amazonaws.com") # Execute tool via MCP result = await self.mcp_client.call_tool( name="aws-cost-analyzer", arguments=kwargs ) return result # ADK Agent setup agent = Agent( name="AWS Cost Assistant", model="gemini-2.0-flash-exp", tools=[AWSMCPTool()], instructions="You are an AWS cost optimization assistant..." )

# Multi-agent setup with MCP from google.adk.a2a import AgentCommunication # Cost Analysis Agent cost_agent = Agent( name="Cost Analyzer", tools=[AWSCostMCPTool()], instructions="Analyze AWS costs and identify optimization opportunities" ) # Resource Optimization Agent optimize_agent = Agent( name="Resource Optimizer", tools=[AWSResourceMCPTool()], instructions="Implement cost optimization recommendations" ) # A2A Communication comm = AgentCommunication([cost_agent, optimize_agent]) # Workflow async def cost_optimization_workflow(query): # Step 1: Analyze costs analysis = await cost_agent.process(query) # Step 2: Get optimization recommendations recommendations = await optimize_agent.process(analysis) return recommendations

// CDK Infrastructure import  as cdk from 'aws-cdk-lib'; import  as lambda from 'aws-cdk-lib/aws-lambda'; import  as apigateway from 'aws-cdk-lib/aws-apigateway'; export class MCPServerStack extends cdk.Stack { constructor(scope: Construct, id: string, props?: cdk.StackProps) { super(scope, id, props); // Lambda function for MCP server const mcpServer = new lambda.Function(this, 'MCPServer', { runtime: lambda.Runtime.PYTHON_3_11, handler: 'mcp_server.lambda_handler', code: lambda.Code.fromAsset('src'), timeout: cdk.Duration.minutes(15), memorySize: 1024, environment: { LOG_LEVEL: 'INFO' } }); // API Gateway for HTTP access const api = new apigateway.RestApi(this, 'MCPServerAPI', { restApiName: 'MCP Server Service' }); const integration = new apigateway.LambdaIntegration(mcpServer); api.root.addMethod('POST', integration); } }

2. Fargate MCP Server with High Availability