Skip to content

OpenTech-Lab/aws-mcp-sererless

Folders and files

NameName
Last commit message
Last commit date

Latest commit

Β 

History

1 Commit
Β 
Β 
Β 
Β 
Β 
Β 
Β 
Β 

Repository files navigation

AWS MCP Serverless - Lambda Deployment Guide

This guide shows how to deploy the AWS MCP Server as an AWS Lambda function with a REST API endpoint.(We use Documentation in this README).

🎯 Final Result

  • Lambda Function: aws-docs-search
  • API Endpoint: https://{CREATED_BY_AWS}
  • Real AWS Documentation: Returns actual AWS docs, not mock data
  • Supports Questions: Works with natural language queries

πŸ“‹ Prerequisites

  1. AWS CLI installed and configured:

    aws --version
    aws configure  # Set up your credentials
  2. Python 3.12+ with uv (for local testing):

    uv --version

πŸš€ Step-by-Step Deployment

Step 1: Create the Lambda Function Code

Create lambda_function.py that uses the real AWS Documentation Search API:

import json
import logging
import uuid
import urllib3
from typing import Dict, Any

# Set up logging
logger = logging.getLogger()
logger.setLevel(logging.INFO)

def lambda_handler(event: Dict[str, Any], context) -> Dict[str, Any]:
    """AWS Lambda handler for AWS Documentation search API."""
    try:
        # Parse input parameters
        http_method = event.get('httpMethod', 'GET')
        query_params = event.get('queryStringParameters') or {}

        # Handle Function URL format
        if 'requestContext' in event and 'http' in event['requestContext']:
            http_method = event['requestContext']['http']['method']

        if http_method == 'GET':
            query = query_params.get('q', '')
            limit = int(query_params.get('limit', 5))
        elif http_method == 'POST':
            body = json.loads(event.get('body', '{}'))
            query = body.get('query', '')
            limit = int(body.get('limit', 5))
        else:
            return create_response(405, {'error': 'Method not allowed'})

        # Validate input
        if not query:
            return create_response(400, {'error': 'Query parameter q is required'})

        if limit < 1 or limit > 20:
            limit = 5

        # Search using urllib3 (no async needed)
        results = search_aws_docs_sync(query, limit)

        # Format response
        response_data = {
            'query': query,
            'limit': limit,
            'results_count': len(results),
            'results': results,
            'source': 'Real AWS Documentation Search API'
        }

        return create_response(200, response_data)

    except Exception as e:
        logger.error(f'Lambda function error: {str(e)}', exc_info=True)
        return create_response(500, {
            'error': 'Internal server error',
            'message': str(e)
        })

def search_aws_docs_sync(query: str, limit: int = 5) -> list:
    """Search AWS documentation using urllib3 (synchronous)."""
    try:
        session_id = str(uuid.uuid4())
        search_url = f"https://proxy.search.docs.aws.amazon.com/search?session={session_id}"

        # Exact payload format from working MCP server
        payload = {
            'textQuery': {
                'input': query,
            },
            'contextAttributes': [{'key': 'domain', 'value': 'docs.aws.amazon.com'}],
            'acceptSuggestionBody': 'RawText',
            'locales': ['en_us'],
        }

        headers = {
            'Content-Type': 'application/json',
            'User-Agent': 'AWS-Docs-MCP-Server/1.0',
            'X-MCP-Session-Id': session_id,
        }

        logger.info(f'Searching AWS docs for: {query}')

        # Use urllib3 PoolManager
        http = urllib3.PoolManager()

        response = http.request(
            'POST',
            search_url,
            body=json.dumps(payload),
            headers=headers,
            timeout=30.0
        )

        if response.status == 200:
            data = json.loads(response.data.decode('utf-8'))
            results = []

            # Process suggestions using exact format from working MCP server
            if 'suggestions' in data and data['suggestions']:
                for i, suggestion in enumerate(data['suggestions'][:limit]):
                    if 'textExcerptSuggestion' in suggestion:
                        text_suggestion = suggestion['textExcerptSuggestion']
                        context = None

                        # Extract context using MCP server logic
                        metadata = text_suggestion.get('metadata', {})
                        if 'seo_abstract' in metadata:
                            context = metadata['seo_abstract']
                        elif 'abstract' in metadata:
                            context = metadata['abstract']
                        elif 'summary' in text_suggestion:
                            context = text_suggestion['summary']
                        elif 'suggestionBody' in text_suggestion:
                            context = text_suggestion['suggestionBody']

                        result = {
                            'rank': i + 1,
                            'title': text_suggestion.get('title', ''),
                            'url': text_suggestion.get('link', ''),
                            'context': context or ''
                        }
                        results.append(result)

                logger.info(f'Found {len(results)} real results from AWS API')
            else:
                logger.warning('No results found in AWS API response')

            return results
        else:
            logger.error(f'AWS API returned status {response.status}')
            return [{
                'rank': 1,
                'title': f'AWS API Error for "{query}"',
                'url': '',
                'context': f'HTTP {response.status}: {response.data.decode("utf-8")[:100]}'
            }]

    except Exception as e:
        logger.error(f'AWS search error: {str(e)}')
        return [{
            'rank': 1,
            'title': f'Search Error for "{query}"',
            'url': '',
            'context': f'Error: {str(e)}'
        }]

def create_response(status_code: int, body: dict) -> dict:
    """Create a properly formatted Lambda response."""
    return {
        'statusCode': status_code,
        'headers': {
            'Content-Type': 'application/json',
            'Access-Control-Allow-Origin': '*',
            'Access-Control-Allow-Methods': 'GET, POST, OPTIONS',
            'Access-Control-Allow-Headers': 'Content-Type'
        },
        'body': json.dumps(body, indent=2)
    }

Step 2: Create IAM Role for Lambda

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

Step 3: Create Deployment Package

# Create deployment package (no external dependencies needed - urllib3 is built-in)
zip aws-docs-lambda.zip lambda_function.py

Step 4: Create Lambda Function

# Create the Lambda function
aws lambda create-function \
    --function-name aws-docs-search \
    --runtime python3.12 \
    --role arn:aws:iam::YOUR_ACCOUNT_ID:role/lambda-execution-role \
    --handler lambda_function.lambda_handler \
    --zip-file fileb://aws-docs-lambda.zip \
    --timeout 30 \
    --memory-size 512 \
    --environment Variables='{FASTMCP_LOG_LEVEL=ERROR,AWS_DOCUMENTATION_PARTITION=aws}'

Step 5: Create Function URL (Public API)

# Create Function URL for HTTP access
aws lambda create-function-url-config \
    --function-name aws-docs-search \
    --cors 'AllowCredentials=false,AllowHeaders=*,AllowMethods=*,AllowOrigins=*,MaxAge=86400' \
    --auth-type NONE

Step 6: Add Public Access Permission

# Add permission for public Function URL access
aws lambda add-permission \
    --function-name aws-docs-search \
    --statement-id FunctionURLAllowPublicAccess \
    --action lambda:InvokeFunctionUrl \
    --principal "*" \
    --function-url-auth-type NONE

πŸ§ͺ Testing Your API

GET Request:

curl "https://YOUR_FUNCTION_URL/?q=EC2+instance+types&limit=3"

POST Request:

curl -X POST "https://YOUR_FUNCTION_URL/" \
  -H "Content-Type: application/json" \
  -d '{"query": "Amazon Bedrock getting started", "limit": 3}'

Natural Language Questions:

curl "https://YOUR_FUNCTION_URL/?q=how+to+launch+EC2+instance&limit=3"

πŸ“Š Example Response

{
  "query": "EC2 instance types",
  "limit": 2,
  "results_count": 2,
  "results": [
    {
      "rank": 1,
      "title": "Amazon EC2 instance types - Amazon Elastic Compute Cloud",
      "url": "https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/instance-types.html",
      "context": "Amazon EC2 instances offer varying compute, memory, storage capabilities optimized for different use cases..."
    },
    {
      "rank": 2,
      "title": "Amazon EC2 instance types - AWS Elastic Beanstalk",
      "url": "https://docs.aws.amazon.com/elasticbeanstalk/latest/dg/using-features.managing.ec2.instance-types.html",
      "context": "Elastic Beanstalk supports Amazon EC2 instance types based on AWS Graviton arm64 architecture..."
    }
  ],
  "source": "Real AWS Documentation Search API"
}

πŸ”§ Key Technical Details

Why This Works:

  1. Uses urllib3 (built into Python) - no external dependencies
  2. Exact API format from the working MCP server
  3. Proper response parsing for AWS's suggestions format
  4. Synchronous HTTP - no async/await complications in Lambda

Required IAM Policies:

  • Lambda Role: AWSLambdaBasicExecutionRole
  • Function URL: lambda:InvokeFunctionUrl permission

API Format Details:

  • Request payload: Uses textQuery structure with contextAttributes
  • Response parsing: Processes suggestions with textExcerptSuggestion
  • Context extraction: Prioritizes seo_abstract β†’ abstract β†’ summary β†’ suggestionBody

πŸ’° Cost Considerations

  • Lambda: Free tier includes 1M requests/month
  • Function URL: No additional cost
  • Data transfer: Minimal for documentation search
  • Total estimated cost: < $1/month for moderate usage

🚨 Common Issues & Solutions

Issue: "Unable to import module"

Solution: Use only built-in Python libraries (urllib3, json, uuid)

Issue: "Running with asyncio requires installation"

Solution: Use synchronous HTTP calls with urllib3, not httpx

Issue: "Invalid request body" (400 error)

Solution: Use exact payload format from working MCP server

Issue: Function URL returns "Forbidden"

Solution: Add lambda:InvokeFunctionUrl permission

πŸ”„ Update Function Code

# Update function code
zip aws-docs-lambda-updated.zip lambda_function.py
aws lambda update-function-code \
    --function-name aws-docs-search \
    --zip-file fileb://aws-docs-lambda-updated.zip

πŸ—‘οΈ Cleanup

# Delete Function URL
aws lambda delete-function-url-config --function-name aws-docs-search

# Delete Lambda function
aws lambda delete-function --function-name aws-docs-search

# Delete IAM role
aws iam detach-role-policy --role-name lambda-execution-role --policy-arn arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
aws iam delete-role --role-name lambda-execution-role

βœ… Success Indicators

  • βœ… Function returns real AWS documentation URLs (not mock data)
  • βœ… Context contains actual AWS service descriptions
  • βœ… Supports both keyword searches and natural language questions
  • βœ… Response includes "source": "Real AWS Documentation Search API"
  • βœ… Returns proper HTTP status codes and CORS headers

πŸŽ‰ Congratulations! You now have a fully functional AWS Documentation search API powered by the real AWS Documentation Search service, deployed as a serverless Lambda function.

About

No description, website, or topics provided.

Resources

Stars

Watchers

Forks

Releases

No releases published

Packages

No packages published

Languages