Lambda Patterns

<!-- AUTO-GENERATED by export-skills.py — DO NOT EDIT -->
---
name: lambda-patterns
description: AWS Lambda patterns for AI agent tool execution. Use when building Bedrock Agent action groups, serverless API backends for agent tools, or event-driven AI processing pipelines.
tags: [aws, lambda, serverless, agents]
---

# AWS Lambda Patterns for AI Agent Tools

Design serverless functions that serve as the execution layer for AI agent action groups.

## When to Use

- Building Bedrock Agent action groups backed by Lambda
- Creating serverless API endpoints for agent tool calls
- Event-driven document processing (S3 trigger -> Lambda -> Bedrock)
- Wrapping enterprise APIs (Salesforce, ServiceNow, SAP) as Lambda-based tools

## Patterns

### 1. Bedrock Agent Action Group Handler

```python
import json
import os
import boto3

def lambda_handler(event, context):
    """Standard Bedrock Agent action group Lambda handler."""
    action_group = event.get("actionGroup")
    api_path = event.get("apiPath")
    http_method = event.get("httpMethod")
    parameters = {p["name"]: p["value"] for p in event.get("parameters", [])}
    body = event.get("requestBody", {}).get("content", {}).get(
        "application/json", {}
    ).get("properties", [])
    body_dict = {p["name"]: p["value"] for p in body} if body else {}

    # Route to appropriate handler
    handlers = {
        ("GET", "/tickets"): list_tickets,
        ("GET", "/tickets/{ticketId}"): get_ticket,
        ("POST", "/tickets"): create_ticket,
        ("PUT", "/tickets/{ticketId}"): update_ticket,
    }

    handler = handlers.get((http_method, api_path))
    if handler:
        result = handler(parameters, body_dict)
        status_code = 200
    else:
        result = {"error": f"Unknown: {http_method} {api_path}"}
        status_code = 404

    return {
        "messageVersion": "1.0",
        "response": {
            "actionGroup": action_group,
            "apiPath": api_path,
            "httpMethod": http_method,
            "httpStatusCode": status_code,
            "responseBody": {
                "application/json": {"body": json.dumps(result)}
            },
        },
    }
```

### 2. Enterprise API Wrapper (e.g., ServiceNow)

```python
import requests
import os

SERVICENOW_INSTANCE = os.environ["SERVICENOW_INSTANCE"]
SERVICENOW_USER = os.environ["SERVICENOW_USER"]
SERVICENOW_PASSWORD = os.environ["SERVICENOW_PASSWORD"]

def create_incident(params: dict, body: dict) -> dict:
    """Create a ServiceNow incident from agent request."""
    url = f"https://{SERVICENOW_INSTANCE}.service-now.com/api/now/table/incident"
    payload = {
        "short_description": body.get("description"),
        "urgency": body.get("urgency", "3"),
        "impact": body.get("impact", "3"),
        "category": body.get("category", "inquiry"),
        "caller_id": body.get("callerId"),
    }
    response = requests.post(
        url, json=payload,
        auth=(SERVICENOW_USER, SERVICENOW_PASSWORD),
        headers={"Content-Type": "application/json", "Accept": "application/json"},
    )
    response.raise_for_status()
    result = response.json()["result"]
    return {
        "incidentNumber": result["number"],
        "sysId": result["sys_id"],
        "state": result["state"],
    }
```

### 3. Event-Driven Document Processing

```python
import boto3
import json

bedrock = boto3.client("bedrock-runtime")

def process_s3_document(event, context):
    """Triggered by S3 upload -- extract and summarize document."""
    s3 = boto3.client("s3")
    for record in event["Records"]:
        bucket = record["s3"]["bucket"]["name"]
        key = record["s3"]["object"]["key"]

        # Read document
        obj = s3.get_object(Bucket=bucket, Key=key)
        content = obj["Body"].read().decode("utf-8")

        # Summarize with Claude
        response = bedrock.invoke_model(
            modelId="anthropic.claude-3-5-sonnet-20241022-v2:0",
            body=json.dumps({
                "anthropic_version": "bedrock-2023-05-31",
                "max_tokens": 1024,
                "messages": [{
                    "role": "user",
                    "content": f"Summarize this document:\n\n{content[:50000]}"
                }]
            }),
        )
        summary = json.loads(response["body"].read())

        # Store summary back to S3
        s3.put_object(
            Bucket=bucket,
            Key=f"summaries/{key}.summary.json",
            Body=json.dumps({"source": key, "summary": summary}),
        )
```

### 4. Lambda Layer for Shared Dependencies

```bash
# Create shared layer for enterprise API clients
mkdir -p python/lib/python3.12/site-packages
pip install requests boto3 -t python/lib/python3.12/site-packages/
zip -r enterprise-tools-layer.zip python/
aws lambda publish-layer-version \
  --layer-name enterprise-tools \
  --zip-file fileb://enterprise-tools-layer.zip \
  --compatible-runtimes python3.12
```

## Anti-Patterns

- Putting secrets in environment variables directly -- use AWS Secrets Manager with caching
- Cold start-heavy functions for real-time agent tools -- use provisioned concurrency or SnapStart
- Monolithic Lambda handling all action groups -- split by domain for independent scaling
- Synchronous calls to slow enterprise APIs without timeout -- set appropriate timeouts and retries
- Missing error handling -- Bedrock Agents need structured error responses

## References

- [Bedrock Agent Action Groups](https://docs.aws.amazon.com/bedrock/latest/userguide/agents-action-group.html)
- [Lambda Best Practices](https://docs.aws.amazon.com/lambda/latest/dg/best-practices.html)
- [AWS Lambda MCP Server](https://github.com/awslabs/mcp)

<!-- Source: .faos/custom/skills/cloud/aws/lambda-patterns/SKILL.md -->