Merge pull request #2990 from pjdavis-aws/pjdavis-aws-feature-appsync-events-lambda-agentcore-cdk

New serverless pattern - appsync-events-lambda-agentcore

Author: julianwood

appsync-events-lambda-agentcore-cdk/.gitignore

@@ -0,0 +1,15 @@
+*.swp
+package-lock.json
+__pycache__
+.pytest_cache
+.venv
+*.egg-info
+.DS_Store
+
+# CDK asset staging directory
+.cdk.staging
+cdk.out
+
+# Dev tooling
+.kiro
+mise.local.toml

appsync-events-lambda-agentcore-cdk/.pylintrc

@@ -0,0 +1,2 @@
+[format]
+max-line-length=150

appsync-events-lambda-agentcore-cdk/README.md

@@ -0,0 +1,176 @@
+# AWS AppSync Events integration with AWS Lambda and Amazon Bedrock AgentCore
+
+This pattern deploys a real-time streaming chat service using AWS AppSync Events with AWS Lambda to invoke a Strands agent running on Amazon Bedrock AgentCore Runtime.
+
+Learn more about this pattern at Serverless Land Patterns: https://serverlessland.com/patterns/appsync-events-lambda-agentcore-cdk
+
+Important: this application uses various AWS services and there are costs associated with these services after the Free Tier usage - please see the [AWS Pricing page](https://aws.amazon.com/pricing/) for details. You are responsible for any AWS costs incurred. No warranty is implied in this example.
+
+## Requirements
+
+* [Create an AWS account](https://portal.aws.amazon.com/gp/aws/developer/registration/index.html) if you do not already have one and log in. The IAM user that you use must have sufficient permissions to make necessary AWS service calls and manage AWS resources.
+* [AWS CLI installed and configured](https://docs.aws.amazon.com/cli/latest/userguide/install-cliv2.html)
+* [Git installed](https://git-scm.com/book/en/v2/Getting-Started-Installing-Git)
+* [Python 3.14](https://www.python.org/downloads/) with [pip](https://pip.pypa.io/en/stable/installation/)
+* [Node.js 22](https://nodejs.org/en/download/)
+* [AWS CDK v2](https://docs.aws.amazon.com/cdk/v2/guide/getting_started.html) (`npm install -g aws-cdk`)
+* [Finch](https://runfinch.com/) or [Docker](https://docs.docker.com/get-docker/) (used for CDK bundling)
+
+## Deployment Instructions
+
+1. Create a new directory, navigate to that directory in a terminal and clone the GitHub repository:
+    ```
+    git clone https://github.com/aws-samples/serverless-patterns
+    ```
+1. Change directory to the pattern directory:
+    ```
+    cd appsync-events-lambda-agentcore-cdk
+    ```
+1. Create and activate a Python virtual environment:
+    ```
+    python -m venv .venv
+    source .venv/bin/activate        # On Windows: .venv\Scripts\activate
+    ```
+1. Install Python dependencies:
+    ```
+    pip install -r requirements.txt
+    ```
+1. Set your target AWS region (must be a region where [Bedrock AgentCore](https://docs.aws.amazon.com/general/latest/gr/bedrock_agentcore.html) is available):
+    ```
+    export AWS_REGION=eu-west-1     # On Windows: set AWS_REGION=eu-west-1
+    ```
+    ```
+    export AWS_REGION=eu-west-1      # On Windows: set AWS_REGION=eu-west-1
+    ```
+1. If you are using [Finch](https://runfinch.com/) instead of Docker, set the `CDK_DOCKER` environment variable:
+    ```
+    export CDK_DOCKER=finch          # On Windows: set CDK_DOCKER=finch
+    ```
+1. Bootstrap CDK in your account/region (if not already done):
+    ```
+    cdk bootstrap
+    ```
+1. Deploy the stack:
+    ```
+    cdk deploy
+    ```
+1. Note the outputs from the CDK deployment process. These contain the AppSync Events HTTP endpoint, WebSocket endpoint, and API key needed for testing.
+
+## How it works
+
+![Architecture diagram](images/architecture.png)
+
+Figure 1 - Architecture
+
+1. The client publishes a message to the inbound channel (`/chat/{conversationId}`) via HTTP POST to AppSync Events.
+2. AppSync Events triggers the agent invoker Lambda function via direct Lambda integration.
+3. The agent invoker validates the payload, invokes the stream relay Lambda asynchronously, and returns immediately. This two-Lambda split is necessary because AppSync invokes the handler synchronously — a long-running stream would block the response.
+4. The stream relay calls `invoke_agent_runtime` on the Bedrock AgentCore Runtime, which hosts a Strands agent container, and consumes the Server-Sent Events (SSE) stream.
+5. The stream relay publishes each chunk back to the response channel on AppSync Events (`/responses/chat/{conversationId}`).
+6. The client receives agent response tokens in real time via the WebSocket subscription.
+
+The client subscribes to the response channel before publishing. Separate channel namespaces (`chat` for inbound, `responses` for outbound) ensure the stream relay's publishes do not re-trigger the agent invoker.
+
+The agent is a Strands-based research assistant with access to `http_request`, `calculator`, and `current_time` tools, backed by S3 session persistence for multi-turn conversations.
+
+## Testing
+
+### Automated tests
+
+Install the test dependencies:
+
+```bash
+pip install -r requirements-dev.txt
+```
+
+Run the tests:
+
+```bash
+pytest tests/unit -v                  # unit tests (no deployed stack needed)
+pytest tests/integration -v -s        # integration tests with streaming output
+```
+
+### Using the AppSync Pub/Sub Editor
+
+You can test the deployed service directly from the AWS Console using the AppSync Events built-in Pub/Sub Editor. No additional tooling required.
+
+1. Open the [AWS AppSync console](https://console.aws.amazon.com/appsync/) in the region you deployed to (e.g. `eu-west-1`).
+1. Select the Event API created by the stack (look for the API with "EventApi" in the name).
+1. Choose the **Pub/Sub Editor** tab.
+1. Scroll to the bottom of the page. The API key is pre-populated in the authorization token field. Choose **Connect** to establish a WebSocket connection.
+1. In the **Subscribe** panel, select `responses` from the namespace dropdown, then enter the path:
+    ```
+    /chat/test-conversation-1
+    ```
+    > **Note:** The namespace is automatically prepended to the path. Since the namespace is `responses`, the full channel becomes `/responses/chat/test-conversation-1`. Do not include the namespace in the path field.
+1. Choose **Subscribe**.
+
+    ![AppSync Pub/Sub Editor — Subscribe panel](images/appsync-pubsub-subscribe.jpg)
+    
+    Figure 2 - AppSync Pub/Sub Editor - Subscribe panel
+
+1. Scroll back to the top of the page to the **Publish** panel. Select `chat` from the namespace dropdown, then enter the path:
+    ```
+    /test-conversation-1
+    ```
+    
+    > **Note:** The namespace `chat` is prepended automatically, so the full channel is `/chat/test-conversation-1`. Do not enter `/chat/test-conversation-1` as the path — that would produce `/chat/chat/test-conversation-1` and messages will not reach the subscriber.
+    
+    Enter this JSON as the event payload:
+    
+    ```json
+    [
+        {
+            "message": "What is 347 multiplied by 29?",
+            "sessionId": "test-conversation-1"
+        }
+    ]
+    ```
+    Choose **Publish**. When prompted, choose **WebSocket** as the publish method.
+
+    ![AppSync Pub/Sub Editor — Publish panel](images/appsync-pubsub-publish.jpg)
+    
+    Figure 3 - AppSync Pub/Sub Editor - Publish panel
+
+1. Scroll back down to the bottom of the page to watch the subscription panel — you should see streaming chunk events arrive in real time, followed by a final completion event containing the full response.
+
+    ![AppSync Pub/Sub Editor — Subscribe results](images/appsync-pubsub-subscribe-result.jpg)
+    
+    Figure 4 - AppSync Pub/Sub Editor - Subscribe results
+
+    The following demo shows the full publish and subscribe flow:
+
+    ![Pub/Sub Editor demo](images/pubsub-demo.gif)
+
+    Figure 5 - AppSync Pub/Sub Editor - Demo
+
+A few things to note:
+
+- The `sessionId` value ties messages to a conversation. Use the same `sessionId` across publishes to test multi-turn conversation with session persistence.
+- The subscribe channel must be prefixed with `/responses` — the agent invoker publishes responses to `/responses/chat/{conversationId}` to avoid re-triggering itself.
+- You can try different prompts to exercise the agent's tools: ask it to fetch a URL (`http_request`), do arithmetic (`calculator`), or tell you the current time (`current_time`).
+
+## Authentication
+
+This example uses an API key for authentication to keep things simple. API keys are suitable for development and testing but are not recommended for production workloads.
+
+AppSync Events supports several authentication methods that are better suited for production:
+
+- **Amazon Cognito user pools** — ideal for end-user authentication in web and mobile apps.
+- **AWS IAM** — best for server-to-server or backend service communication.
+- **OpenID Connect (OIDC)** — use with third-party identity providers.
+- **Lambda authorizers** — for custom authorization logic.
+
+You can configure multiple authorization modes on a single API and apply different modes per channel namespace. See the [AppSync Events authorization and authentication](https://docs.aws.amazon.com/appsync/latest/eventapi/configure-event-api-auth.html) documentation for details.
+
+## Cleanup
+
+1. Delete the stack (answer `y` when prompted to confirm):
+    ```
+    cdk destroy
+    ```
+
+----
+Copyright 2025 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+
+SPDX-License-Identifier: MIT-0

appsync-events-lambda-agentcore-cdk/agents/chat/Dockerfile

@@ -0,0 +1,20 @@
+FROM ghcr.io/astral-sh/uv:python3.14-bookworm-slim
+
+WORKDIR /app
+
+ENV UV_SYSTEM_PYTHON=1 UV_COMPILE_BYTECODE=1
+
+COPY chat/requirements.txt requirements.txt
+RUN uv pip install --system --no-cache -r requirements.txt
+
+ARG AWS_REGION
+ENV AWS_REGION=${AWS_REGION}
+
+RUN useradd -m -u 1000 bedrock_agentcore
+USER bedrock_agentcore
+
+EXPOSE 8080
+
+COPY chat/ /app
+
+CMD ["opentelemetry-instrument", "python", "-m", "entrypoint"]

appsync-events-lambda-agentcore-cdk/agents/chat/entrypoint.py

@@ -0,0 +1,96 @@
+"""Chat agent entrypoint for AgentCore runtime.
+
+Pure streaming agent with S3-backed session persistence.
+Yields response chunks via SSE. Has no knowledge of delivery
+mechanism (AppSync, WebSocket, etc.).
+"""
+
+import os
+import logging
+
+from strands import Agent
+from strands.models import BedrockModel
+from strands.session.s3_session_manager import S3SessionManager
+from strands_tools import http_request, calculator, current_time
+from bedrock_agentcore.runtime import BedrockAgentCoreApp
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+app = BedrockAgentCoreApp()
+
+MODEL_ID = os.environ.get("BEDROCK_MODEL_ID")
+if not MODEL_ID:
+    raise ValueError("BEDROCK_MODEL_ID environment variable is required")
+
+REGION = os.environ.get("AWS_REGION")
+if not REGION:
+    raise ValueError("AWS_REGION environment variable is required")
+SESSION_BUCKET = os.environ.get("SESSION_BUCKET")
+
+SYSTEM_PROMPT = """\
+You are a research assistant with access to the web, a calculator, and a clock.
+
+You can:
+- Fetch and summarise content from any public URL using http_request
+- Perform mathematical calculations using calculator
+- Check the current date and time in any timezone using current_time
+
+When fetching web content, prefer converting HTML to markdown for readability
+by setting convert_to_markdown=true. Always cite the URL you fetched.
+Keep responses clear and concise.
+"""
+
+
+def _create_agent(session_id: str | None = None) -> Agent:
+    """Create a Strands agent with Bedrock model and optional session."""
+    model = BedrockModel(model_id=MODEL_ID, region_name=REGION)
+
+    kwargs = {
+        "system_prompt": SYSTEM_PROMPT,
+        "model": model,
+        "tools": [http_request, calculator, current_time],
+    }
+
+    if session_id and SESSION_BUCKET:
+        kwargs["session_manager"] = S3SessionManager(
+            session_id=session_id,
+            bucket=SESSION_BUCKET,
+            region_name=REGION,
+        )
+
+    return Agent(**kwargs)
+
+
+@app.entrypoint
+async def invoke(payload=None):
+    """Stream agent response as SSE events."""
+    if not payload:
+        yield {"status": "error", "error": "payload is required"}
+        return
+
+    query = payload.get("content") or payload.get("prompt")
+    if not query:
+        yield {"status": "error", "error": "content or prompt is required"}
+        return
+
+    session_id = payload.get("sessionId")
+    logger.info("Processing query: %s (session: %s)", query[:100], session_id)
+
+    agent = _create_agent(session_id)
+
+    async for event in agent.stream_async(query):
+        if "data" in event:
+            yield {"data": event["data"]}
+        elif "result" in event:
+            result = event["result"]
+            yield {
+                "result": {
+                    "stop_reason": str(result.stop_reason),
+                    "message": result.message,
+                },
+            }
+
+
+if __name__ == "__main__":
+    app.run()

appsync-events-lambda-agentcore-cdk/agents/chat/requirements.txt

@@ -0,0 +1,4 @@
+strands-agents>=1.29.0
+strands-agents-tools>=0.2.22
+bedrock-agentcore>=1.4.4
+aws-opentelemetry-distro>=0.15.0

appsync-events-lambda-agentcore-cdk/app.py

@@ -0,0 +1,27 @@
+#!/usr/bin/env python3
+"""CDK app entrypoint for the AppSync Events + Lambda + AgentCore stack."""
+import os
+
+import aws_cdk as cdk
+from cdk_nag import AwsSolutionsChecks
+
+from cdk.stack import ChatStack
+
+
+app = cdk.App()
+
+stack_name = app.node.try_get_context("stack_name") or "AppsyncLambdaAgentcore"
+
+region = os.environ.get("AWS_REGION")
+if not region:
+    raise EnvironmentError("AWS_REGION environment variable must be set")
+
+ChatStack(
+    app,
+    stack_name,
+    env=cdk.Environment(region=region),
+)
+
+cdk.Aspects.of(app).add(AwsSolutionsChecks(verbose=True))
+
+app.synth()