# Custom integration for each tool
def connect_to_crm():
    # Custom CRM code
    pass

def connect_to_database():
    # Custom DB code
    pass

def connect_to_api():
    # Custom API code
    pass

# Single standard interface
mcp_client.discover_tools()
mcp_client.call_tool("query_crm", params)
mcp_client.call_tool("query_database", params)
mcp_client.call_tool("call_api", params)

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "query_database",
    "arguments": {
      "query": "SELECT * FROM orders WHERE id = 12345"
    }
  }
}

# MCP Tool Definition
@mcp.tool()
def send_email(to: str, subject: str, body: str) -> str:
    """Send an email to a recipient"""
    # Send email logic
    return f"Email sent to {to}"

# AI can call this tool
result = mcp_client.call_tool("send_email", {
    "to": "user@example.com",
    "subject": "Order Confirmation",
    "body": "Your order has shipped"
})

# MCP Resource Definition
@mcp.resource("company://policies/return-policy")
def get_return_policy() -> str:
    """Get the company return policy"""
    return load_policy_document()

# AI can read this resource
policy = mcp_client.read_resource("company://policies/return-policy")

# MCP Prompt Definition
@mcp.prompt()
def code_review_prompt(code: str, language: str) -> str:
    """Generate a code review prompt"""
    return f"""
    Review this {language} code for:
    - Bugs and errors
    - Performance issues
    - Security vulnerabilities
    - Best practices

    Code:
    {code}
    """

# AI can use this prompt
prompt = mcp_client.get_prompt("code_review_prompt", {
    "code": user_code,
    "language": "python"
})

# Bedrock Agent can use MCP servers as tools
from langchain_aws import ChatBedrock
import mcp

# Initialize Bedrock model
bedrock_llm = ChatBedrock(
    model_id="anthropic.claude-3-sonnet-20240229-v1:0"
)

# Connect to MCP server
mcp_client = mcp.Client("stdio", command="python", args=["server.py"])

# Discover tools from MCP server
tools = mcp_client.list_tools()

# Bedrock agent uses MCP tools
for tool in tools:
    print(f"Available: {tool.name} - {tool.description}")

# Agent execution with MCP tools
response = bedrock_llm.invoke(
    "Query the database for customer orders",
    tools=tools
)

# Expose Bedrock KB as MCP resource
from mcp.server import Server
import boto3

mcp_server = Server("bedrock-kb-server")

@mcp_server.resource("bedrock://kb/{kb_id}")
def get_kb_data(kb_id: str, query: str):
    """Query Bedrock Knowledge Base"""
    client = boto3.client('bedrock-agent-runtime')

    response = client.retrieve(
        knowledgeBaseId=kb_id,
        retrievalQuery={'text': query}
    )

    return response['retrievalResults']

# Amazon Q can leverage MCP servers
# Example: Whiteboard to cloud workflow

# 1. User uploads whiteboard image to Amazon Q
# 2. Q uses Bedrock Data Automation MCP server
# 3. MCP server processes image, extracts architecture
# 4. Q generates CloudFormation templates
# 5. Q deploys to AWS

# MCP Server for Bedrock Data Automation
@mcp.tool()
def extract_architecture(image_data: bytes) -> dict:
    """Extract architecture from whiteboard image"""
    client = boto3.client('bedrock-data-automation')

    response = client.invoke_data_automation_async(
        inputConfig={'s3Uri': upload_to_s3(image_data)},
        outputConfig={'s3Uri': 's3://output-bucket/'},
        dataAutomationProjectArn='arn:aws:...'
    )

    return parse_architecture(response)

# Tools provided by AgentCore MCP Server
tools = [
    "create_agent",           # Create new Bedrock agent
    "update_agent",           # Update agent configuration
    "create_action_group",    # Add action groups
    "create_knowledge_base",  # Create KB
    "test_agent",            # Test agent functionality
    "deploy_agent"           # Deploy to production
]

# Example usage
@mcp.tool()
def create_bedrock_agent(name: str, instructions: str):
    """Create a new Bedrock agent"""
    # Automated agent creation
    pass

# Tools for data automation
tools = [
    "extract_text",          # Extract text from images
    "parse_documents",       # Parse structured documents
    "extract_tables",        # Extract tables
    "classify_content"       # Classify document types
]

# Example
@mcp.tool()
def extract_whiteboard_architecture(image_url: str):
    """Extract architecture diagram from whiteboard"""
    # Uses Bedrock Data Automation
    pass

# API Gateway MCP Proxy Configuration
Resources:
  MCPProxy:
    Type: AWS::ApiGateway::RestApi
    Properties:
      Name: OrdersAPIMCP
      MCPEnabled: true
      MCPConfiguration:
        ToolMapping:
          - Path: /orders/{id}
            Method: GET
            ToolName: query_order
            Description: "Get order by ID"
          - Path: /orders
            Method: POST
            ToolName: create_order
            Description: "Create new order"

# MCP Server for S3 operations
from mcp.server import Server
import boto3

s3_server = Server("s3-mcp-server")
s3_client = boto3.client('s3')

@s3_server.tool()
def list_buckets() -> list:
    """List all S3 buckets"""
    response = s3_client.list_buckets()
    return [b['Name'] for b in response['Buckets']]

@s3_server.tool()
def upload_file(bucket: str, key: str, content: str):
    """Upload file to S3"""
    s3_client.put_object(
        Bucket=bucket,
        Key=key,
        Body=content.encode()
    )
    return f"Uploaded to s3://{bucket}/{key}"

@s3_server.resource("s3://{bucket}/{key}")
def read_file(bucket: str, key: str) -> str:
    """Read file from S3"""
    response = s3_client.get_object(Bucket=bucket, Key=key)
    return response['Body'].read().decode()

# MCP Server for DynamoDB
from mcp.server import Server
import boto3

dynamodb_server = Server("dynamodb-mcp-server")
dynamodb = boto3.resource('dynamodb')

@dynamodb_server.tool()
def query_table(table_name: str, key: dict) -> dict:
    """Query DynamoDB table"""
    table = dynamodb.Table(table_name)
    response = table.get_item(Key=key)
    return response.get('Item', {})

@dynamodb_server.tool()
def put_item(table_name: str, item: dict):
    """Put item in DynamoDB"""
    table = dynamodb.Table(table_name)
    table.put_item(Item=item)
    return f"Item added to {table_name}"

@dynamodb_server.resource("dynamodb://{table_name}/schema")
def get_table_schema(table_name: str) -> dict:
    """Get DynamoDB table schema"""
    table = dynamodb.Table(table_name)
    return {
        'TableName': table.table_name,
        'KeySchema': table.key_schema,
        'AttributeDefinitions': table.attribute_definitions
    }

# MCP Server for Lambda invocation
from mcp.server import Server
import boto3
import json

lambda_server = Server("lambda-mcp-server")
lambda_client = boto3.client('lambda')

@lambda_server.tool()
def invoke_function(function_name: str, payload: dict) -> dict:
    """Invoke Lambda function"""
    response = lambda_client.invoke(
        FunctionName=function_name,
        InvocationType='RequestResponse',
        Payload=json.dumps(payload)
    )

    result = json.loads(response['Payload'].read())
    return result

@lambda_server.tool()
def list_functions() -> list:
    """List all Lambda functions"""
    response = lambda_client.list_functions()
    return [f['FunctionName'] for f in response['Functions']]

@lambda_server.resource("lambda://{function_name}/config")
def get_function_config(function_name: str) -> dict:
    """Get Lambda function configuration"""
    response = lambda_client.get_function_configuration(
        FunctionName=function_name
    )
    return response

# MCP Server for CloudWatch Logs
from mcp.server import Server
import boto3

cloudwatch_server = Server("cloudwatch-mcp-server")
logs_client = boto3.client('logs')

@cloudwatch_server.tool()
def query_logs(log_group: str, query: str, hours: int = 1) -> list:
    """Query CloudWatch Logs"""
    import time

    start_time = int((time.time() - hours * 3600) * 1000)
    end_time = int(time.time() * 1000)

    response = logs_client.start_query(
        logGroupName=log_group,
        startTime=start_time,
        endTime=end_time,
        queryString=query
    )

    query_id = response['queryId']

    # Wait for query to complete
    while True:
        result = logs_client.get_query_results(queryId=query_id)
        if result['status'] == 'Complete':
            return result['results']
        time.sleep(1)

@cloudwatch_server.resource("cloudwatch://logs/{log_group}")
def get_recent_logs(log_group: str, limit: int = 100) -> list:
    """Get recent log events"""
    response = logs_client.filter_log_events(
        logGroupName=log_group,
        limit=limit
    )
    return response['events']

# install: pip install mcp
from mcp.server import Server
from mcp.types import Tool, Resource, TextContent

# Create server
server = Server("my-aws-server")

# Define a tool
@server.tool()
def greet(name: str) -> str:
    """Greet a user by name"""
    return f"Hello, {name}!"

# Define a resource
@server.resource("config://app-settings")
def get_settings() -> dict:
    """Get application settings"""
    return {
        "region": "us-east-1",
        "environment": "production"
    }

# Define a prompt
@server.prompt()
def analysis_prompt(data: str) -> str:
    """Generate analysis prompt"""
    return f"Analyze this data and provide insights:\n{data}"

# Run server
if __name__ == "__main__":
    server.run()

# Complete AWS MCP Server Example
from mcp.server import Server
import boto3
import json

server = Server("aws-services-server")

# AWS Clients
s3 = boto3.client('s3')
dynamodb = boto3.resource('dynamodb')
bedrock = boto3.client('bedrock-runtime')

# S3 Tools
@server.tool()
def s3_list_objects(bucket: str, prefix: str = "") -> list:
    """List objects in S3 bucket"""
    response = s3.list_objects_v2(Bucket=bucket, Prefix=prefix)
    return [obj['Key'] for obj in response.get('Contents', [])]

@server.tool()
def s3_read_file(bucket: str, key: str) -> str:
    """Read file from S3"""
    response = s3.get_object(Bucket=bucket, Key=key)
    return response['Body'].read().decode('utf-8')

# DynamoDB Tools
@server.tool()
def dynamodb_query(table: str, key_name: str, key_value: str) -> dict:
    """Query DynamoDB table"""
    table_resource = dynamodb.Table(table)
    response = table_resource.get_item(Key={key_name: key_value})
    return response.get('Item', {})

# Bedrock Tools
@server.tool()
def bedrock_invoke(prompt: str, model_id: str = "anthropic.claude-3-sonnet-20240229-v1:0") -> str:
    """Invoke Bedrock model"""
    body = json.dumps({
        "anthropic_version": "bedrock-2023-05-31",
        "max_tokens": 4096,
        "messages": [{"role": "user", "content": prompt}]
    })

    response = bedrock.invoke_model(
        modelId=model_id,
        body=body
    )

    result = json.loads(response['body'].read())
    return result['content'][0]['text']

# Resources
@server.resource("aws://account-info")
def get_account_info() -> dict:
    """Get AWS account information"""
    sts = boto3.client('sts')
    identity = sts.get_caller_identity()
    return {
        'account_id': identity['Account'],
        'user_arn': identity['Arn']
    }

# Run server
if __name__ == "__main__":
    import asyncio
    asyncio.run(server.run())

# MCP Server as Lambda function
import json
from mcp.server import Server

server = Server("lambda-mcp-server")

@server.tool()
def process_data(data: str) -> str:
    """Process data"""
    return f"Processed: {data}"

def lambda_handler(event, context):
    """Lambda handler for MCP requests"""
    # Parse MCP request
    mcp_request = json.loads(event['body'])

    # Handle MCP protocol
    if mcp_request['method'] == 'tools/list':
        return {
            'statusCode': 200,
            'body': json.dumps(server.list_tools())
        }
    elif mcp_request['method'] == 'tools/call':
        result = server.call_tool(
            mcp_request['params']['name'],
            mcp_request['params']['arguments']
        )
        return {
            'statusCode': 200,
            'body': json.dumps(result)
        }

# Dockerfile for MCP Server
FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY server.py .

EXPOSE 8000

CMD ["python", "server.py"]

# ECS Task Definition
Resources:
  MCPServerTask:
    Type: AWS::ECS::TaskDefinition
    Properties:
      Family: mcp-server
      NetworkMode: awsvpc
      RequiresCompatibilities:
        - FARGATE
      Cpu: 256
      Memory: 512
      ContainerDefinitions:
        - Name: mcp-server
          Image: !Sub ${AWS::AccountId}.dkr.ecr.${AWS::Region}.amazonaws.com/mcp-server:latest
          PortMappings:
            - ContainerPort: 8000
              Protocol: tcp

#!/bin/bash
# User data script for EC2

# Install dependencies
yum update -y
yum install -y python3 python3-pip

# Install MCP server
pip3 install mcp boto3

# Download server code
aws s3 cp s3://my-bucket/mcp-server.py /opt/mcp-server.py

# Run server
python3 /opt/mcp-server.py

# MCP Server for DevOps
@server.tool()
def check_service_health(service_name: str) -> dict:
    """Check health of AWS service"""
    # Check CloudWatch metrics
    # Check service status
    # Return health report
    pass

@server.tool()
def deploy_application(app_name: str, version: str):
    """Deploy application to AWS"""
    # Trigger CodePipeline
    # Monitor deployment
    # Return status
    pass

@server.tool()
def scale_service(service_name: str, desired_count: int):
    """Scale ECS service"""
    # Update ECS service
    # Monitor scaling
    pass

# Usage with Bedrock
user: "Check health of my web service and scale if needed"
ai: Uses check_service_health → Sees high CPU → Uses scale_service

# MCP Server for Data Analysis
@server.tool()
def query_athena(query: str) -> list:
    """Query data using Athena"""
    athena = boto3.client('athena')
    # Execute query
    # Return results
    pass

@server.tool()
def analyze_with_bedrock(data: str) -> str:
    """Analyze data using Bedrock"""
    bedrock = boto3.client('bedrock-runtime')
    # Invoke model for analysis
    pass

@server.resource("s3://data-lake/{dataset}")
def get_dataset(dataset: str) -> dict:
    """Get dataset from S3"""
    # Read from S3
    # Return data
    pass

# Workflow
user: "Analyze sales data from last quarter"
ai: 
  1. Uses get_dataset to fetch data
  2. Uses query_athena to aggregate
  3. Uses analyze_with_bedrock for insights

# MCP Server for Customer Support
@server.tool()
def query_customer_data(customer_id: str) -> dict:
    """Get customer information from DynamoDB"""
    table = dynamodb.Table('customers')
    return table.get_item(Key={'id': customer_id})['Item']

@server.tool()
def query_order_status(order_id: str) -> dict:
    """Get order status"""
    # Query orders table
    pass

@server.tool()
def create_support_ticket(customer_id: str, issue: str) -> str:
    """Create support ticket"""
    # Create ticket in system
    pass

@server.resource("bedrock://kb/support-docs")
def get_support_docs(query: str) -> list:
    """Query support documentation from Bedrock KB"""
    kb_client = boto3.client('bedrock-agent-runtime')
    response = kb_client.retrieve(
        knowledgeBaseId='KB123',
        retrievalQuery={'text': query}
    )
    return response['retrievalResults']

# Interaction
user: "Where is my order #12345?"
ai:
  1. Uses query_order_status
  2. Finds order shipped
  3. Responds with tracking info

# MCP Server for Code Deployment
@server.tool()
def generate_lambda_function(description: str) -> str:
    """Generate Lambda function code using Bedrock"""
    bedrock = boto3.client('bedrock-runtime')
    prompt = f"Generate Python Lambda function: {description}"
    # Generate code
    return generated_code

@server.tool()
def deploy_lambda(function_name: str, code: str):
    """Deploy Lambda function"""
    lambda_client = boto3.client('lambda')
    # Create/update function
    # Return ARN
    pass

@server.tool()
def create_api_gateway(lambda_arn: str) -> str:
    """Create API Gateway for Lambda"""
    apigw = boto3.client('apigatewayv2')
    # Create API
    # Create integration
    # Return API URL
    pass

# Workflow
user: "Create a Lambda function that processes S3 uploads and expose it via API"
ai:
  1. Uses generate_lambda_function
  2. Uses deploy_lambda
  3. Uses create_api_gateway
  4. Returns API endpoint

# MCP Server for Multi-Account Management
@server.tool()
def assume_role(account_id: str, role_name: str) -> dict:
    """Assume role in another account"""
    sts = boto3.client('sts')
    response = sts.assume_role(
        RoleArn=f"arn:aws:iam::{account_id}:role/{role_name}",
        RoleSessionName="MCPSession"
    )
    return response['Credentials']

@server.tool()
def list_resources_cross_account(accounts: list, resource_type: str) -> dict:
    """List resources across multiple accounts"""
    results = {}
    for account in accounts:
        creds = assume_role(account, 'ReadOnlyRole')
        # List resources with assumed credentials
        results[account] = resources
    return results

@server.tool()
def deploy_cross_account(accounts: list, template: str):
    """Deploy CloudFormation across accounts"""
    for account in accounts:
        creds = assume_role(account, 'DeployRole')
        # Deploy template
    pass

# openapi.yaml
openapi: 3.0.0
info:
  title: User API
  version: 1.0.0

paths:
  /users/{userId}:
    get:
      summary: Get user by ID
      operationId: getUser
      parameters:
        - name: userId
          in: path
          required: true
          schema:
            type: string
      responses:
        '200':
          description: User found
          content:
            application/json:
              schema:
                type: object
                properties:
                  id:
                    type: string
                  name:
                    type: string
                  email:
                    type: string

# Auto-generated from OpenAPI spec
from mcp.server import Server
import requests

server = Server("user-api-mcp")
BASE_URL = "https://api.example.com"

@server.tool()
def get_user(userId: str) -> dict:
    """
    Get user by ID

    Generated from: GET /users/{userId}
    """
    response = requests.get(f"{BASE_URL}/users/{userId}")
    response.raise_for_status()
    return response.json()

# Install
pip install automcp

# Generate MCP server from OpenAPI spec
automcp generate openapi.yaml --output server.py

# Run generated server
python server.py

from fastmcp import FastMCP

# Create MCP server from OpenAPI spec
mcp = FastMCP.from_openapi("openapi.yaml")

# Customize if needed
@mcp.tool()
def custom_tool():
    """Add custom tools alongside OpenAPI tools"""
    pass

# Run
mcp.run()

# CloudFormation template
Resources:
  MyRestApi:
    Type: AWS::ApiGateway::RestApi
    Properties:
      Name: OrdersAPI
      Description: Orders REST API

  MCPIntegration:
    Type: AWS::ApiGateway::MCPIntegration
    Properties:
      RestApiId: !Ref MyRestApi
      MCPConfiguration:
        Enabled: true
        AutoGenerateTools: true
        ToolNamingStrategy: operationId
        AuthenticationMethod: IAM

# AI agent using API Gateway MCP Proxy
from langchain_aws import ChatBedrock
import mcp

# Connect to API Gateway MCP endpoint
mcp_client = mcp.Client(
    "https://api.example.com/mcp",
    auth={"type": "aws-iam"}
)

# Discover tools (automatically from OpenAPI)
tools = mcp_client.list_tools()

# Use with Bedrock
bedrock = ChatBedrock(model_id="anthropic.claude-3-sonnet")
response = bedrock.invoke(
    "Get order #12345",
    tools=tools
)

# ✅ Good: Clear operation IDs become tool names
paths:
  /users/{id}:
    get:
      operationId: getUserById  # Becomes: get_user_by_id tool

# ❌ Bad: Missing operation ID
paths:
  /users/{id}:
    get:
      # No operationId - generator creates generic name

# ✅ Good: Detailed descriptions help AI understand
paths:
  /orders:
    post:
      summary: Create a new order
      description: |
        Creates a new order in the system. Requires customer ID,
        product IDs, and shipping address. Returns order ID and
        estimated delivery date.
      operationId: createOrder

# ✅ Good: Use IAM roles and policies
@server.tool()
def secure_operation():
    # Use IAM role attached to Lambda/EC2
    client = boto3.client('s3')  # Uses IAM role automatically

# ❌ Bad: Hardcode credentials
@server.tool()
def insecure_operation():
    client = boto3.client(
        's3',
        aws_access_key_id='AKIAIOSFODNN7EXAMPLE',  # Never do this!
        aws_secret_access_key='wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY'
    )

# ✅ Good: Comprehensive error handling
@server.tool()
def robust_operation(bucket: str, key: str) -> dict:
    """Read from S3 with error handling"""
    try:
        s3 = boto3.client('s3')
        response = s3.get_object(Bucket=bucket, Key=key)
        return {
            'success': True,
            'data': response['Body'].read().decode()
        }
    except s3.exceptions.NoSuchKey:
        return {'success': False, 'error': 'File not found'}
    except s3.exceptions.NoSuchBucket:
        return {'success': False, 'error': 'Bucket not found'}
    except Exception as e:
        return {'success': False, 'error': str(e)}

# ✅ Good: Cache expensive operations
from functools import lru_cache

@lru_cache(maxsize=100)
def get_table_schema(table_name: str):
    """Cached table schema lookup"""
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table(table_name)
    return table.key_schema

# ✅ Good: Use pagination for large results
@server.tool()
def list_all_objects(bucket: str) -> list:
    """List all S3 objects with pagination"""
    s3 = boto3.client('s3')
    paginator = s3.get_paginator('list_objects_v2')

    all_objects = []
    for page in paginator.paginate(Bucket=bucket):
        all_objects.extend(page.get('Contents', []))

    return all_objects

# ✅ Good: Comprehensive logging
import logging

logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

@server.tool()
def monitored_operation(param: str) -> dict:
    """Operation with logging"""
    logger.info(f"Starting operation with param: {param}")

    try:
        result = perform_operation(param)
        logger.info(f"Operation successful: {result}")
        return {'success': True, 'result': result}
    except Exception as e:
        logger.error(f"Operation failed: {str(e)}", exc_info=True)
        return {'success': False, 'error': str(e)}

# Send logs to CloudWatch
cloudwatch_handler = watchtower.CloudWatchLogHandler()
logger.addHandler(cloudwatch_handler)

# ✅ Good: Use context managers
@server.tool()
def process_large_file(bucket: str, key: str):
    """Process large file efficiently"""
    s3 = boto3.client('s3')

    # Stream file instead of loading into memory
    response = s3.get_object(Bucket=bucket, Key=key)

    with response['Body'] as stream:
        for line in stream.iter_lines():
            process_line(line)

    return "Processing complete"

# ✅ Good: Clean up resources
@server.tool()
def temporary_resource_operation():
    """Create and clean up temporary resources"""
    s3 = boto3.client('s3')
    temp_bucket = create_temp_bucket()

    try:
        # Use temporary bucket
        result = perform_operation(temp_bucket)
        return result
    finally:
        # Always clean up
        delete_bucket(temp_bucket)

# ✅ Good: Clear, detailed documentation
@server.tool()
def well_documented_tool(
    bucket: str,
    prefix: str = "",
    max_keys: int = 1000
) -> list:
    """
    List objects in an S3 bucket with optional filtering.

    Args:
        bucket: Name of the S3 bucket
        prefix: Optional prefix to filter objects (default: "")
        max_keys: Maximum number of keys to return (default: 1000)

    Returns:
        List of object keys matching the criteria

    Raises:
        NoSuchBucket: If the bucket doesn't exist
        AccessDenied: If lacking permissions

    Example:
        >>> list_objects("my-bucket", prefix="logs/", max_keys=100)
        ['logs/2025-01-01.log', 'logs/2025-01-02.log']
    """
    s3 = boto3.client('s3')
    response = s3.list_objects_v2(
        Bucket=bucket,
        Prefix=prefix,
        MaxKeys=max_keys
    )
    return [obj['Key'] for obj in response.get('Contents', [])]

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "s3:GetObject",
        "s3:ListBucket"
      ],
      "Resource": [
        "arn:aws:s3:::my-bucket",
        "arn:aws:s3:::my-bucket/*"
      ]
    }
  ]
}

MCPServerRole:
  Type: AWS::IAM::Role
  Properties:
    AssumeRolePolicyDocument:
      Version: '2012-10-17'
      Statement:
        - Effect: Allow
          Principal:
            Service: lambda.amazonaws.com
          Action: sts:AssumeRole
    ManagedPolicyArns:
      - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
    Policies:
      - PolicyName: MCPServerPolicy
        PolicyDocument:
          Version: '2012-10-17'
          Statement:
            - Effect: Allow
              Action:
                - s3:GetObject
                - dynamodb:GetItem
                - bedrock:InvokeModel
              Resource: '*'

# Implement authentication for MCP server
from functools import wraps
import jwt

def require_auth(f):
    @wraps(f)
    def decorated_function(*args, **kwargs):
        token = get_token_from_request()

        try:
            # Verify JWT token
            payload = jwt.decode(token, SECRET_KEY, algorithms=['HS256'])

            # Check permissions
            if not has_permission(payload['user_id'], f.__name__):
                raise PermissionError("Insufficient permissions")

            return f(*args, **kwargs)
        except jwt.InvalidTokenError:
            raise AuthenticationError("Invalid token")

    return decorated_function

@server.tool()
@require_auth
def sensitive_operation(data: str):
    """Protected operation requiring authentication"""
    pass

# Encrypt sensitive data
from cryptography.fernet import Fernet
import boto3

def encrypt_data(data: str) -> str:
    """Encrypt data using KMS"""
    kms = boto3.client('kms')

    response = kms.encrypt(
        KeyId='alias/my-key',
        Plaintext=data.encode()
    )

    return response['CiphertextBlob']

def decrypt_data(encrypted_data: bytes) -> str:
    """Decrypt data using KMS"""
    kms = boto3.client('kms')

    response = kms.decrypt(
        CiphertextBlob=encrypted_data
    )

    return response['Plaintext'].decode()

@server.tool()
def store_sensitive_data(data: str):
    """Store encrypted data in S3"""
    encrypted = encrypt_data(data)
    s3 = boto3.client('s3')
    s3.put_object(
        Bucket='secure-bucket',
        Key='sensitive-data',
        Body=encrypted,
        ServerSideEncryption='aws:kms'
    )

# VPC Configuration for MCP Server
Resources:
  MCPServerSecurityGroup:
    Type: AWS::EC2::SecurityGroup
    Properties:
      GroupDescription: Security group for MCP server
      VpcId: !Ref VPC
      SecurityGroupIngress:
        - IpProtocol: tcp
          FromPort: 443
          ToPort: 443
          SourceSecurityGroupId: !Ref ClientSecurityGroup
      SecurityGroupEgress:
        - IpProtocol: tcp
          FromPort: 443
          ToPort: 443
          DestinationPrefixListId: !Ref S3PrefixList

# Comprehensive audit logging
import json
from datetime import datetime

@server.tool()
def audited_operation(user_id: str, action: str, resource: str):
    """Operation with audit logging"""

    # Log to CloudWatch
    logger.info(json.dumps({
        'timestamp': datetime.utcnow().isoformat(),
        'user_id': user_id,
        'action': action,
        'resource': resource,
        'source_ip': get_source_ip()
    }))

    # Also log to CloudTrail via AWS API call
    cloudtrail = boto3.client('cloudtrail')
    cloudtrail.put_insight_selectors(
        TrailName='mcp-audit-trail',
        InsightSelectors=[{
            'InsightType': 'ApiCallRateInsight'
        }]
    )

    # Perform operation
    result = perform_operation(resource)

    # Log result
    logger.info(f"Operation completed: {result}")

    return result