diff --git a/lambda-microvms-multi-tenant-ai-agents/.gitignore b/lambda-microvms-multi-tenant-ai-agents/.gitignore new file mode 100644 index 000000000..2e18438ff --- /dev/null +++ b/lambda-microvms-multi-tenant-ai-agents/.gitignore @@ -0,0 +1 @@ +.gateway-token diff --git a/lambda-microvms-multi-tenant-ai-agents/README.md b/lambda-microvms-multi-tenant-ai-agents/README.md new file mode 100644 index 000000000..36fa39993 --- /dev/null +++ b/lambda-microvms-multi-tenant-ai-agents/README.md @@ -0,0 +1,105 @@ +# Multi-tenant AI agents on AWS Lambda MicroVMs + +This pattern runs a self-hosted AI agent ([OpenClaw](https://github.com/openclaw/openclaw)) **one isolated Lambda MicroVM per tenant**, with per-tenant state persisted on Amazon EFS, model calls served by Amazon Bedrock through a VPC endpoint, and an orchestrator Lambda behind API Gateway that cold-starts, resumes, and reaps tenant VMs on demand. + +Most AI agents sit idle most of the time, yet an "always-on" deployment bills 24/7. Lambda MicroVMs flip that model: an idle tenant's VM auto-suspends (barely billed), a fully idle tenant is terminated with its state parked on EFS for ≈$0, and a returning tenant resumes from a Firecracker snapshot in seconds — memory intact. Each tenant gets a dedicated micro-VM, so isolation is a hard security boundary by design, and the workload obtains AWS credentials from the MicroVM's IMDSv2 execution role with no static keys. + +Learn more about this pattern at Serverless Land Patterns: << Add the live URL here >> + +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 v2 >= 2.35](https://docs.aws.amazon.com/cli/latest/userguide/install-cliv2.html) installed and configured — must include the `lambda-microvms` and `lambda-core` subcommands (check with `aws lambda-microvms help`) +* [Git Installed](https://git-scm.com/book/en/v2/Getting-Started-Installing-Git) +* Python 3 and [uv](https://docs.astral.sh/uv/) (or a recent pip), plus the `zip` and `openssl` utilities +* A [Lambda MicroVMs launch region](https://docs.aws.amazon.com/lambda/latest/dg/lambda-microvms.html) (us-east-1, us-east-2, us-west-2, eu-west-1, ap-northeast-1) +* Amazon Bedrock **model access for Anthropic Claude** enabled in the target region + +Docker is **not** required locally — the container image build runs on AWS as part of the `AWS::Lambda::MicrovmImage` resource. + +## 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 serverless-patterns/lambda-microvms-multi-tenant-ai-agents + ``` +1. Deploy the whole system with one command (takes ~10 minutes: CloudFormation stack + server-side MicroVM image build + VPC egress connector): + ``` + ./deploy.sh + # e.g. ./deploy.sh openclaw-mt us-east-1 + ``` + The script pre-flights the CLI and target region, uploads two zip artifacts to S3 (the MicroVM image source and the bundled orchestrator Lambda — both under content-hashed keys, so a code change always reaches AWS and an unchanged redeploy is a true no-op) and then runs a single `aws cloudformation deploy`. Everything else — VPC with NAT internet egress, EFS, Bedrock VPC endpoints (runtime + control plane), IAM roles, DynamoDB tenant registry, the MicroVM image (built server-side by CloudFormation via `AWS::Lambda::MicrovmImage`), the VPC egress connector, the orchestrator Lambda, API Gateway, and the EventBridge sweeper — is declared in `template.yaml`. A random per-checkout gateway token is minted into `.gateway-token` (override with `$GATEWAY_TOKEN`). +1. Register a tenant in the DynamoDB registry: + ``` + ./add-tenant.sh tenant1 + ``` +1. Note the `ApiEndpoint` output printed at the end of the deploy. Tenant webhook URLs take the form `/tg/`. + +## How it works + +![Architecture](images/architecture.png) + +1. **Image build**: CloudFormation creates the `AWS::Lambda::MicrovmImage` resource. Lambda downloads the zip artifact from S3, executes the Dockerfile server-side (installs the OpenClaw agent, a sidecar, an EFS mount daemon, and a persistent gateway bridge), and takes a Firecracker snapshot. +2. **Message routing**: a message for a tenant arrives at API Gateway and invokes the orchestrator Lambda (router role). The router ACKs immediately and hands the message to an async worker invocation. +3. **Tenant lookup**: the worker checks the DynamoDB tenant registry. If the tenant's MicroVM is alive (RUNNING or SUSPENDED — suspended VMs auto-resume in seconds), the turn is forwarded. If the tenant is cold, the worker calls `run-microvm` (guarded by a conditional-write lock so concurrent messages launch only one VM), waits for the VM to mount the tenant's EFS subdirectory, and then forwards the turn. +4. **Agent turn**: inside the MicroVM, a sidecar receives the turn and passes it to the OpenClaw gateway over a persistent WebSocket bridge — the gateway holds the agent state warm in memory, so a turn takes ~2 seconds instead of re-reading state over NFS on every message. If a long-running turn outlives the worker invocation, the worker relays polling to a fresh async self-invocation, so a turn is bounded by the VM's 8-hour lifetime rather than Lambda's 15 minutes. +5. **Model call**: the agent calls Amazon Bedrock through VPC interface endpoints (runtime for inference, control plane for live model discovery at cold start). General internet egress for the agent's web search/fetch tools goes through a NAT gateway, while Bedrock and EFS traffic stays on the private VPCE/mount-target paths. Credentials come from the MicroVM's IMDSv2 execution role — no static keys anywhere. +6. **State persistence**: the tenant's full agent state (config + conversation memory) lives under `/tenants/` on EFS. It survives suspend, resume, termination, and the MicroVM's 8-hour maximum lifetime — a relaunched VM adopts the existing state and the agent remembers everything. +7. **Lifecycle sweep**: an EventBridge rule invokes the orchestrator (sweeper role) every 10 minutes to terminate VMs idle beyond the threshold and reconcile the registry against ground truth. Tenants flow hot (RUNNING) → warm (SUSPENDED) → cold (TERMINATED, state on EFS) automatically. + +## Testing + +Chat with a tenant synchronously (bypasses API Gateway's 30s timeout so you can watch a cold start, which takes ~90 seconds; subsequent warm turns take ~2 seconds): + +```bash +./chat.sh tenant1 "Remember my lucky number is 7777." +# → cold: True | reply: (agent confirms) + +./chat.sh tenant1 "What's my lucky number?" +# → cold: False | reply: 7777 +``` + +To verify cross-generation persistence, terminate the tenant's MicroVM (`aws lambda-microvms terminate-microvm ...` or simply wait for the sweeper), then ask again — the relaunched VM adopts the EFS state and still answers `7777` with `cold: True`. + +Tenant isolation: register a second tenant and confirm it cannot see the first tenant's memory: + +```bash +./add-tenant.sh tenant2 +./chat.sh tenant2 "What's my lucky number?" +# → the agent does not know — tenant2 has its own EFS subdirectory and its own VM +``` + +### Optional: Telegram push front-end + +Each tenant can be bound to its own Telegram bot. Create a bot with [@BotFather](https://t.me/BotFather) (`/newbot` → token), then register the tenant with the token and a webhook secret of your choosing: + +```bash +./add-tenant.sh tenant3 +``` + +The script sets the bot's webhook to `/tg/tenant3`. Messaging the bot then drives the same router → worker → MicroVM flow, and replies are delivered back through the Telegram Bot API. Use a dedicated bot per tenant — registering overwrites the bot's existing webhook. + +Over Telegram the worker additionally provides: + +* **Streaming replies** — a placeholder message that grows via `editMessageText` while the model generates, with a `▌` cursor until the final edit. +* **Images** — send a photo with or without a caption; the worker pulls it from Telegram, ships it into the VM as a base64 attachment, and the agent answers about what it sees. +* **`/model` switching** — e.g. `/model amazon-bedrock/us.anthropic.claude-sonnet-5`, `/model default` to reset. The model catalog is discovered live from Bedrock at each cold start, so newly launched models are switchable without a redeploy. + +## Cleanup + +```bash +./teardown.sh +``` + +The script terminates this stack's running MicroVMs (matched by image ARN, so VMs from other stacks are untouched), deletes the CloudFormation stack (which removes the MicroVM image, network connector, EFS, VPC, IAM roles, DynamoDB table, Lambda, and API Gateway), and finally empties and deletes the artifact bucket. + +---- +Copyright 2026 Amazon.com, Inc. or its affiliates. All Rights Reserved. + +SPDX-License-Identifier: MIT-0 diff --git a/lambda-microvms-multi-tenant-ai-agents/add-tenant.sh b/lambda-microvms-multi-tenant-ai-agents/add-tenant.sh new file mode 100755 index 000000000..3396fae50 --- /dev/null +++ b/lambda-microvms-multi-tenant-ai-agents/add-tenant.sh @@ -0,0 +1,38 @@ +#!/usr/bin/env bash +# Register a tenant. For Telegram push, also pass a bot token + secret and this +# script sets the webhook to /tg/. +# Usage: ./add-tenant.sh STACK REGION TENANT_ID [BOT_TOKEN] [WEBHOOK_SECRET] +set -euo pipefail +if [ $# -lt 3 ]; then + echo "usage: $0 STACK REGION TENANT_ID [BOT_TOKEN] [WEBHOOK_SECRET]" >&2; exit 1 +fi +STACK="$1"; REGION="$2"; TID="$3" +BOT="${4:-}"; SECRET="${5:-}" +API="$(aws cloudformation describe-stacks --region "$REGION" --stack-name "$STACK" \ + --query "Stacks[0].Outputs[?OutputKey=='ApiEndpoint'].OutputValue" --output text)" + +ITEM="{\"tenantId\":{\"S\":\"$TID\"},\"state\":{\"S\":\"COLD\"},\"generation\":{\"N\":\"0\"}" +[ -n "$BOT" ] && ITEM="$ITEM,\"botToken\":{\"S\":\"$BOT\"}" +[ -n "$SECRET" ] && ITEM="$ITEM,\"webhookSecret\":{\"S\":\"$SECRET\"}" +ITEM="$ITEM}" +aws dynamodb put-item --region "$REGION" --table-name "${STACK}-tenants" --item "$ITEM" +echo "registered tenant '$TID' (state=COLD)" + +if [ -n "$BOT" ] && [ -n "$SECRET" ]; then + # Parse Telegram's response instead of dumping raw JSON: a bad bot token must + # fail loudly (non-zero), not scroll past as {"ok":false,...}. + curl -s "https://api.telegram.org/bot${BOT}/setWebhook" \ + --data-urlencode "url=${API}/tg/${TID}" \ + --data-urlencode "secret_token=${SECRET}" \ + --data-urlencode "drop_pending_updates=true" | python3 -c ' +import json, sys +d = json.load(sys.stdin) +if d.get("ok"): + print("setWebhook: ok") +else: + print("ERROR: setWebhook failed:", d.get("description", d), file=sys.stderr) + print(" Check the bot token (from @BotFather) and try again.", file=sys.stderr) + sys.exit(1) +' + echo "webhook -> ${API}/tg/${TID}" +fi diff --git a/lambda-microvms-multi-tenant-ai-agents/chat.sh b/lambda-microvms-multi-tenant-ai-agents/chat.sh new file mode 100755 index 000000000..5435aa449 --- /dev/null +++ b/lambda-microvms-multi-tenant-ai-agents/chat.sh @@ -0,0 +1,25 @@ +#!/usr/bin/env bash +# Synchronous test chat for a tenant (cold-starts the VM if needed). Invokes the +# orchestrator worker directly so it isn't bound by API Gateway's 30s timeout — +# handy for validating cold starts (~90s) from a terminal. +# Usage: ./chat.sh STACK REGION TENANT_ID "message" [sessionKey] +set -euo pipefail +if [ $# -lt 4 ]; then + echo "usage: $0 STACK REGION TENANT_ID \"message\" [sessionKey]" >&2; exit 1 +fi +STACK="$1"; REGION="$2"; TID="$3"; MSG="$4"; SESS="${5:-cli}" +FN="${STACK}-orchestrator" + +# Heads-up before the blocking invoke: a cold tenant means a ~90s MicroVM cold start, +# which would otherwise look like a silent hang. +STATE="$(aws dynamodb get-item --region "$REGION" --table-name "${STACK}-tenants" \ + --key "{\"tenantId\":{\"S\":\"$TID\"}}" --query 'Item.state.S' --output text 2>/dev/null || true)" +if [ "$STATE" != "RUNNING" ]; then + echo "Tenant is cold — starting MicroVM, first turn takes ~90s ..." +fi +PAYLOAD="$(python3 -c 'import json,sys; print(json.dumps({"_worker":{"tenantId":sys.argv[1],"update":{"message":{"chat":{"id":sys.argv[2]},"text":sys.argv[3]}}}}))' "$TID" "$SESS" "$MSG")" +OUT="$(mktemp)" +aws lambda invoke --region "$REGION" --function-name "$FN" \ + --cli-binary-format raw-in-base64-out --payload "$PAYLOAD" \ + --cli-read-timeout 300 "$OUT" >/dev/null +python3 -c 'import json,sys; d=json.load(open(sys.argv[1])); print("cold:",d.get("cold"),"| reply:",d.get("reply"))' "$OUT" diff --git a/lambda-microvms-multi-tenant-ai-agents/deploy.sh b/lambda-microvms-multi-tenant-ai-agents/deploy.sh new file mode 100755 index 000000000..e906a38ec --- /dev/null +++ b/lambda-microvms-multi-tenant-ai-agents/deploy.sh @@ -0,0 +1,143 @@ +#!/usr/bin/env bash +# End-to-end deploy for multi-tenant OpenClaw on Lambda MicroVMs. +# +# CloudFormation does almost everything — including the MicroVM image and the VPC egress +# connector, which ARE native CFN resources (AWS::Lambda::MicrovmImage / +# AWS::Lambda::NetworkConnector). The only thing CFN can't do is put *bytes* in S3, so this +# script's imperative part is just: ensure a bucket + upload the two zips. Then one +# `cloudformation deploy` builds the image, provisions the connector, and wires the +# orchestrator Lambda via GetAtt. +# +# Prereqs: AWS CLI v2 >= 2.35 (has the lambda-microvms models), python3, uv (or a modern +# pip), zip. Credentials for a MicroVMs launch region with Bedrock Claude access. +# +# Usage: ./deploy.sh [STACK_NAME] [REGION] +set -euo pipefail + +STACK="${1:-openclaw-mt}" +REGION="${2:-us-east-1}" +HERE="$(cd "$(dirname "$0")" && pwd)" + +say(){ printf '\n\033[1;36m== %s ==\033[0m\n' "$*"; } + +# ---------- 0. Pre-flight (fail fast, before anything is created) ---------- +say "0/4 Pre-flight checks" +if ! aws lambda-microvms help >/dev/null 2>&1; then + echo "ERROR: this AWS CLI has no 'lambda-microvms' subcommands (needs AWS CLI v2 >= 2.35)." + echo " installed: $(aws --version 2>&1)" + exit 1 +fi +# Cheap read against the target region: catches not-yet-launched regions (the launch +# list keeps growing — probe instead of hardcoding it) and missing credentials. +if ! ERR="$(aws lambda-microvms list-microvms --region "$REGION" 2>&1 >/dev/null)"; then + echo "ERROR: Lambda MicroVMs isn't reachable in region '$REGION':" + printf ' %s\n' "$ERR" + echo " If the service hasn't launched in '$REGION' yet, use a launch region and re-run." + exit 1 +fi +echo " aws cli ok; region '$REGION' ok" + +ACCOUNT="$(aws sts get-caller-identity --query Account --output text)" +# Gateway token: explicit $GATEWAY_TOKEN wins; otherwise reuse .gateway-token, minting +# it once per checkout (random) so redeploys are idempotent and no shared default ships. +TOKEN_FILE="$HERE/.gateway-token" +if [ -z "${GATEWAY_TOKEN:-}" ]; then + [ -f "$TOKEN_FILE" ] || openssl rand -hex 16 > "$TOKEN_FILE" + GATEWAY_TOKEN="$(cat "$TOKEN_FILE")" + TOKEN_NOTE="(kept in ${TOKEN_FILE#$HERE/}; redeploys reuse it)" +else + TOKEN_NOTE="(from \$GATEWAY_TOKEN)" +fi +BUCKET="${STACK}-artifact-${ACCOUNT}-${REGION}" +# S3 keys carry a content hash: CloudFormation only rebuilds the MicroVM image / +# updates the Lambda when a resource PROPERTY changes, so a fixed key means code +# changes silently never reach AWS on redeploy. Hash of the inputs fixes that +# (and makes an unchanged redeploy a true no-op). +IMG_SRC="Dockerfile openclaw.json hooks.py start.sh efs-monitor.sh gw-bridge.cjs materialize-models.mjs" +IMG_HASH="$(cd "$HERE/src/microvm" && cat $IMG_SRC | shasum -a 256 | cut -c1-12)" +IMG_KEY="microvm-images/openclaw-${IMG_HASH}.zip" +CODE_HASH="$(shasum -a 256 "$HERE/src/orchestrator/handler.py" | cut -c1-12)" +CODE_KEY="lambda/orchestrator-${CODE_HASH}.zip" + +# ---------- 1. Artifact bucket (the one imperative prerequisite) ---------- +say "1/4 Ensure artifact bucket: $BUCKET" +if ! aws s3api head-bucket --bucket "$BUCKET" 2>/dev/null; then + if [ "$REGION" = "us-east-1" ]; then + aws s3api create-bucket --bucket "$BUCKET" --region "$REGION" >/dev/null + else + aws s3api create-bucket --bucket "$BUCKET" --region "$REGION" \ + --create-bucket-configuration LocationConstraint="$REGION" >/dev/null + fi + aws s3api put-public-access-block --bucket "$BUCKET" \ + --public-access-block-configuration BlockPublicAcls=true,BlockPublicPolicy=true,IgnorePublicAcls=true,RestrictPublicBuckets=true +fi + +# ---------- 2. MicroVM image artifact (Dockerfile + app) ---------- +say "2/4 Package & upload MicroVM image artifact ($IMG_KEY)" +IMGZIP="$(mktemp -d)/microvm.zip" +( cd "$HERE/src/microvm" && zip -j -q "$IMGZIP" $IMG_SRC ) +aws s3 cp "$IMGZIP" "s3://${BUCKET}/${IMG_KEY}" --region "$REGION" + +# ---------- 3. Orchestrator Lambda zip (boto3 + lambda-microvms model overlay + handler) ---------- +say "3/4 Bundle & upload orchestrator Lambda" +PKG="$(mktemp -d)" +if command -v uv >/dev/null 2>&1; then + uv pip install --quiet --python "$(command -v python3)" --target "$PKG" boto3 +else + python3 -m pip install --quiet --upgrade --target "$PKG" boto3 \ + || { echo "ERROR: install uv (https://docs.astral.sh/uv/) or a recent pip"; exit 1; } +fi +# PyPI boto3 may lag the lambda-microvms service model; overlay it from the local AWS CLI's +# botocore data (the CLI ships it — that's how this very script calls the service). +CLI_DATA="$(python3 - "$(readlink "$(command -v aws)" || command -v aws)" <<'PY' +import os,sys,glob +root=os.path.realpath(sys.argv[1]) +for _ in range(6): + root=os.path.dirname(root) + hit=glob.glob(os.path.join(root,'**','botocore','data','lambda-microvms'),recursive=True) + if hit: print(os.path.dirname(hit[0])); break +PY +)" +if [ -n "$CLI_DATA" ] && [ -d "$PKG/botocore/data" ]; then + cp -R "$CLI_DATA/lambda-microvms" "$PKG/botocore/data/" 2>/dev/null || true + cp -R "$CLI_DATA/lambda-core" "$PKG/botocore/data/" 2>/dev/null || true +fi +# Hard gate: the bundle MUST contain the model, or the orchestrator only fails at invoke +# time with an unrelated-looking UnknownServiceError. Fail the deploy here instead. +if ! python3 -c "import sys;sys.path.insert(0,'$PKG');import boto3;assert 'lambda-microvms' in boto3.session.Session().get_available_services();print(' boto3',boto3.__version__,'+ lambda-microvms model')" 2>/dev/null; then + echo "ERROR: couldn't get the 'lambda-microvms' service model into the Lambda bundle." + echo " Bundled boto3 lacks it, and extracting it from the local AWS CLI failed" + echo " (CLI data dir found: '${CLI_DATA:-none}')." + echo " Fix: install AWS CLI v2 >= 2.35 via the official installer, or upgrade" + echo " boto3 on PyPI to a version that ships the lambda-microvms model." + exit 1 +fi +cp "$HERE/src/orchestrator/handler.py" "$PKG/" +CODEZIP="$(mktemp -d)/orchestrator.zip"; ( cd "$PKG" && zip -q -r "$CODEZIP" . ) +echo " code key: $CODE_KEY" +aws s3 cp "$CODEZIP" "s3://${BUCKET}/${CODE_KEY}" --region "$REGION" + +# ---------- 4. One CloudFormation deploy: builds image, connector, everything, wired ---------- +say "4/4 CloudFormation deploy (builds MicroVM image ~5min + connector + all infra)" +aws cloudformation deploy --region "$REGION" --stack-name "$STACK" \ + --template-file "$HERE/template.yaml" \ + --capabilities CAPABILITY_NAMED_IAM \ + --parameter-overrides \ + ProjectName="$STACK" \ + GatewayToken="$GATEWAY_TOKEN" \ + ArtifactBucketName="$BUCKET" \ + MicrovmImageKey="$IMG_KEY" \ + OrchestratorCodeKey="$CODE_KEY" + +API="$(aws cloudformation describe-stacks --region "$REGION" --stack-name "$STACK" \ + --query "Stacks[0].Outputs[?OutputKey=='ApiEndpoint'].OutputValue" --output text)" + +say "DONE" +cat < [telegramBotToken] [webhookSecret] + Test (HTTP) : ./chat.sh ${STACK} ${REGION} "your message" + Teardown : ./teardown.sh ${STACK} ${REGION} +EOF diff --git a/lambda-microvms-multi-tenant-ai-agents/example-pattern.json b/lambda-microvms-multi-tenant-ai-agents/example-pattern.json new file mode 100644 index 000000000..b58cb7e58 --- /dev/null +++ b/lambda-microvms-multi-tenant-ai-agents/example-pattern.json @@ -0,0 +1,72 @@ +{ + "title": "Multi-tenant AI agents on AWS Lambda MicroVMs", + "description": "Run a self-hosted AI agent one isolated Lambda MicroVM per tenant, with per-tenant state on Amazon EFS, Amazon Bedrock via a VPC endpoint, and an orchestrator that cold-starts, resumes, and reaps tenant VMs on demand.", + "language": "Python", + "level": "300", + "framework": "AWS CLI", + "introBox": { + "headline": "How it works", + "text": [ + "This pattern deploys a complete multi-tenant control plane for self-hosted AI agents on AWS Lambda MicroVMs. Each tenant gets a dedicated Firecracker micro-VM running the agent, so tenant isolation is a hard security boundary by design rather than a shared runtime with guardrails.", + "The entire system is declared in one CloudFormation template: a VPC with an Amazon EFS file system, Bedrock VPC endpoints (runtime for inference and control plane for live model discovery), and NAT internet egress for the agent's web tools, the MicroVM image itself (built server-side by CloudFormation via the AWS::Lambda::MicrovmImage resource — no local Docker required), a VPC egress network connector (AWS::Lambda::NetworkConnector), all IAM roles, a DynamoDB tenant registry, an orchestrator Lambda behind an API Gateway HTTP API, and an EventBridge sweeper rule. The deploy script only uploads two zip artifacts to S3 (under content-hashed keys, so redeploys reliably pick up code changes) and runs a single cloudformation deploy.", + "When a message for a tenant arrives, the orchestrator's router acknowledges immediately and hands off to an async worker. The worker looks the tenant up in the registry: if the tenant's MicroVM is alive it forwards the turn (a suspended VM auto-resumes from its Firecracker snapshot in seconds, memory intact); if the tenant is cold it launches a fresh MicroVM under a conditional-write lock, waits for the VM to mount the tenant's own subdirectory on EFS, and then forwards the turn. Inside the VM, a sidecar passes the turn to the agent gateway over a persistent WebSocket bridge, which keeps agent state warm in memory — a warm turn completes in about two seconds instead of re-reading state over NFS on every message.", + "The agent calls Amazon Bedrock through the VPC interface endpoint, authenticating with temporary credentials from the MicroVM's IMDSv2 execution role — no static keys are baked into the image or environment. The available model catalog is discovered live from Bedrock at each cold start, so newly launched models are usable without a redeploy. Every tenant's full agent state (configuration and conversation memory) lives on EFS and survives suspend, resume, termination, and the MicroVM's 8-hour maximum lifetime: a relaunched VM adopts the existing state and the agent remembers everything.", + "An EventBridge rule invokes the sweeper every 10 minutes to terminate long-idle VMs and reconcile the registry, so tenants flow hot (RUNNING) to warm (SUSPENDED, barely billed) to cold (TERMINATED, state parked on EFS for near zero cost) automatically. You pay for conversations, not for waiting — the economics that make one-agent-per-tenant viable at scale. Optionally, each tenant can be bound to its own Telegram bot for a push-based chat front-end through the same API Gateway webhook path, with streaming replies, image understanding, and per-tenant model switching." + ] + }, + "gitHub": { + "template": { + "repoURL": "https://github.com/aws-samples/serverless-patterns/tree/main/lambda-microvms-multi-tenant-ai-agents", + "templateURL": "serverless-patterns/lambda-microvms-multi-tenant-ai-agents", + "projectFolder": "lambda-microvms-multi-tenant-ai-agents", + "templateFile": "template.yaml" + } + }, + "resources": { + "bullets": [ + { + "text": "AWS Lambda MicroVMs Documentation", + "link": "https://docs.aws.amazon.com/lambda/latest/dg/lambda-microvms.html" + }, + { + "text": "Firecracker - Secure and fast microVMs for serverless computing", + "link": "https://aws.amazon.com/blogs/opensource/firecracker-open-source-secure-fast-microvm-serverless/" + }, + { + "text": "Amazon Bedrock Documentation", + "link": "https://docs.aws.amazon.com/bedrock/latest/userguide/what-is-bedrock.html" + }, + { + "text": "Amazon EFS Documentation", + "link": "https://docs.aws.amazon.com/efs/latest/ug/whatisefs.html" + } + ] + }, + "deploy": { + "headline": "Deploy the pattern", + "text": [ + "./deploy.sh ", + "./add-tenant.sh tenant1" + ] + }, + "testing": { + "text": [ + "./chat.sh tenant1 \"Remember my lucky number is 7777.\"", + "./chat.sh tenant1 \"What's my lucky number?\"", + "See the README for cross-generation persistence and tenant-isolation tests." + ] + }, + "cleanup": { + "text": [ + "./teardown.sh " + ] + }, + "authors": [ + { + "name": "Shawn Zhang", + "image": "https://avatars.builderprofile.aws.dev/2qK6tUfRW3r2c97Z7X6tCSpr3AQ.webp", + "bio": "Sr. Specialist Solutions Architect, specializing in GenAI & Engineering.", + "linkedin": "hustshawn" + } + ] +} diff --git a/lambda-microvms-multi-tenant-ai-agents/images/architecture.png b/lambda-microvms-multi-tenant-ai-agents/images/architecture.png new file mode 100644 index 000000000..0fae57d48 Binary files /dev/null and b/lambda-microvms-multi-tenant-ai-agents/images/architecture.png differ diff --git a/lambda-microvms-multi-tenant-ai-agents/src/microvm/Dockerfile b/lambda-microvms-multi-tenant-ai-agents/src/microvm/Dockerfile new file mode 100644 index 000000000..c692d1814 --- /dev/null +++ b/lambda-microvms-multi-tenant-ai-agents/src/microvm/Dockerfile @@ -0,0 +1,22 @@ +# OpenClaw on Lambda MicroVMs — EFS-persisted state variant (arm64). +# Runs as ROOT (mounting NFS needs privileges + ALL os capabilities at image level). +FROM ghcr.io/openclaw/openclaw:slim + +USER root +RUN apt-get update && apt-get install -y --no-install-recommends nfs-common \ + && rm -rf /var/lib/apt/lists/* + +COPY openclaw.json hooks.py start.sh efs-monitor.sh gw-bridge.cjs materialize-models.mjs /opt/poc/ +RUN chmod +x /opt/poc/start.sh /opt/poc/efs-monitor.sh \ + && mkdir -p /home/node/.openclaw /mnt/efs \ + && cp /opt/poc/openclaw.json /home/node/.openclaw/openclaw.json \ + && HOME=/home/node node /app/openclaw.mjs plugins install @openclaw/amazon-bedrock-provider \ + && chown -R node:node /home/node \ + # Gateway runs as root; OpenClaw blocks plugins whose files aren't root-owned + # ("suspicious ownership" guard), so the plugin tree must stay root:root. + && chown -R root:root /home/node/.openclaw/npm + +ENV HOME=/home/node +EXPOSE 18789 8080 +ENTRYPOINT ["tini", "-s", "--"] +CMD ["/opt/poc/start.sh"] diff --git a/lambda-microvms-multi-tenant-ai-agents/src/microvm/efs-monitor.sh b/lambda-microvms-multi-tenant-ai-agents/src/microvm/efs-monitor.sh new file mode 100644 index 000000000..f2a5bf0c0 --- /dev/null +++ b/lambda-microvms-multi-tenant-ai-agents/src/microvm/efs-monitor.sh @@ -0,0 +1,85 @@ +#!/usr/bin/env bash +# Tenant-aware persistent EFS mount daemon. +# Stays COMPLETELY quiet during image build (hard-NFS against unreachable IP +# wedges the snapshot). Waits until the orchestrator injects a tenant id via +# the sidecar (POST /tenant -> /var/run/tenant-id), then mounts the EFS and +# binds THIS TENANT's subdir over the state dir, then bounces the gateway. +STATE_DIR=/home/node/.openclaw +EFS_DIR=/mnt/efs +MARKER=/var/run/efs-mounted +TENANT_FILE=/var/run/tenant-id +# Mount target: prefer the EFS DNS name (regional, resolves to the AZ mount-target +# IP inside the VPC) built from EFS_ID; fall back to an explicit EFS_MOUNT_IP. +if [ -n "${EFS_ID:-}" ]; then + EFS_HOST="${EFS_ID}.efs.${AWS_REGION:-us-east-1}.amazonaws.com" +elif [ -n "${EFS_MOUNT_IP:-}" ]; then + EFS_HOST="${EFS_MOUNT_IP}" +else + echo "[efs-monitor] need EFS_ID or EFS_MOUNT_IP"; exit 1 +fi + +mkdir -p "$EFS_DIR" +until [ -s "$TENANT_FILE" ]; do sleep 2; done +TENANT=$(tr -cd 'a-zA-Z0-9_-' < "$TENANT_FILE") +echo "[efs-monitor] tenant '$TENANT' assigned; mount target ${EFS_HOST}; starting mount attempts" + +while true; do + if [ ! -f "$MARKER" ]; then + if timeout 15 mount -t nfs4 -o nfsvers=4.1,rsize=1048576,wsize=1048576,hard,timeo=150,retrans=2 \ + "${EFS_HOST}:/" "$EFS_DIR" 2>/tmp/efs-mount.err; then + echo "[efs-monitor] EFS mounted from ${EFS_HOST} at $(date -u +%FT%TZ)" + TDIR="$EFS_DIR/tenants/$TENANT" + mkdir -p "$TDIR" + if [ ! -f "$TDIR/openclaw.json" ]; then + echo "[efs-monitor] tenant $TENANT first generation: seeding from local state" + cp -a "$STATE_DIR/." "$TDIR/" + else + echo "[efs-monitor] tenant $TENANT has prior state - adopting it" + fi + # Config is image-owned; only agent state persists across generations. Without + # this, a stale EFS openclaw.json shadows every config change shipped in the image. + # Also drop OpenClaw's backups: its watchdog restores .last-good over any config + # that lacks the meta stamp (missing-meta-vs-last-good), undoing the overwrite. + cp /opt/poc/openclaw.json "$TDIR/openclaw.json" + rm -f "$TDIR/openclaw.json.last-good" "$TDIR/openclaw.json.bak" + # Materialize live Bedrock model discovery into the config (~1.3-3.3s once + # per cold start). The gateway's image guard and models list only read the + # explicit models array, never the plugin's live catalog — so we bake the + # discovered catalog (with correct text+image modalities) in here. On any + # failure the static seed list shipped in the image stays as fallback. + node /opt/poc/materialize-models.mjs "$TDIR/openclaw.json" || true + # Plugins must be root-owned or the gateway blocks them (see Dockerfile note); + # fix up trees seeded by older generations. + [ -d "$TDIR/npm" ] && chown -R root:root "$TDIR/npm" 2>/dev/null + # Clear per-session /model pins on cold start. A pin to a since-retired Bedrock + # model fails every turn INCLUDING the /model command that would clear it, + # permanently deadlocking the session. Cold start (reap -> relaunch) is the safe + # reset point: the gateway isn't running against this dir yet. Pins still + # survive suspend/resume; they only reset when the tenant went fully cold. + python3 - "$TDIR/agents/main/sessions/sessions.json" <<'HEAL' 2>/dev/null || true +import json, sys +p = sys.argv[1] +try: + d = json.load(open(p)) +except Exception: + sys.exit(0) +changed = False +for k, e in d.items(): + if not isinstance(e, dict) or not e.get("modelOverride"): + continue + print(f"[efs-monitor] clearing model pin on {k}: {e['modelOverride']}") + for f in ("providerOverride", "modelOverride", "modelOverrideSource", + "model", "modelProvider"): + e.pop(f, None) + changed = True +if changed: + json.dump(d, open(p, "w"), indent=1) +HEAL + mount --bind "$TDIR" "$STATE_DIR" + touch "$MARKER" + echo "[efs-monitor] state dir now EFS-backed for tenant $TENANT; bouncing gateway" + pkill -f "openclaw.mjs gateway" || true + fi + fi + sleep 5 +done diff --git a/lambda-microvms-multi-tenant-ai-agents/src/microvm/gw-bridge.cjs b/lambda-microvms-multi-tenant-ai-agents/src/microvm/gw-bridge.cjs new file mode 100644 index 000000000..ff512108e --- /dev/null +++ b/lambda-microvms-multi-tenant-ai-agents/src/microvm/gw-bridge.cjs @@ -0,0 +1,229 @@ +// Persistent gateway bridge: holds ONE WebSocket to the warm OpenClaw gateway and +// runs agent turns over it — eliminating the ~20s per-message CLI spawn cost. +// Exposes a tiny local HTTP API on 127.0.0.1:8090 that the sidecar calls. +// +// GET /agent?m=&s= -> {"reply": "..."} (sync) +// POST /agent {m,s,attachments} -> {"reply": "..."} (sync, images) +// POST /agent-async {m,s,attachments}-> {"turnId": "..."} (returns at once) +// GET /progress?id= -> {"text","done","reply"?} (poll for stream) +// GET /ready -> {"connected": true} +// +// Frame protocol (reverse-engineered + verified against gateway 2026.6.11): +// server sends event connect.challenge{nonce} +// client sends req connect{minProtocol,maxProtocol,role,scopes,caps,client{mode:backend},auth{token}} +// server res ok -> hello-ok +// client sends req agent{message,sessionKey,idempotencyKey} +// server res ok -> result.payloads[].text (async; deltas stream as events meanwhile) +const http = require("http"); +const crypto = require("node:crypto"); +const WebSocket = require("/app/node_modules/ws"); + +const TOKEN = process.env.OPENCLAW_GATEWAY_TOKEN || "poc-microvm-token-42"; +const GW = "ws://127.0.0.1:18789"; +const PORT = 8090; + +let ws = null; +let connected = false; +const pending = new Map(); // requestId -> {resolve, reject, timer} +// Async turns for the poll-based streaming path: turnId -> {text, done, reply, error, at} +const turns = new Map(); +// Gateway assigns its own runId per run (returned on the ack frame); delta +// events are keyed by that runId, so map it back to our turnId. +const runToTurn = new Map(); +const TURN_TTL_MS = 10 * 60 * 1000; +// Async turns may legitimately run for hours (the Lambda pollers chain across +// invocations); the VM's 8h max lifetime is the real bound, not a poll timeout. +const ASYNC_TURN_TIMEOUT_MS = 8 * 60 * 60 * 1000; + +// GC: drop turns 10 min after their last activity (progress poll or completion). +// An in-flight turn being polled keeps refreshing its timestamp, so only turns +// abandoned by every poller age out. +function gcTurns() { + const cutoff = Date.now() - TURN_TTL_MS; + for (const [id, t] of turns) if (t.at < cutoff) turns.delete(id); + for (const [rid, tid] of runToTurn) if (!turns.has(tid)) runToTurn.delete(rid); +} + +function connect() { + connected = false; + ws = new WebSocket(GW); + ws.on("open", () => {}); + ws.on("message", (buf) => { + let m; + try { m = JSON.parse(buf.toString()); } catch { return; } + if (m.event === "connect.challenge") { + ws.send(JSON.stringify({ + type: "req", id: "connect", method: "connect", + params: { + minProtocol: 4, maxProtocol: 4, role: "operator", + scopes: ["operator.admin"], caps: [], + client: { id: "gateway-client", displayName: "gw-bridge", + version: "2026.6.11", platform: "linux", mode: "backend" }, + auth: { token: TOKEN }, + }, + })); + return; + } + if (m.type === "res" && m.id === "connect") { + connected = !!m.ok; + if (!m.ok) { console.error("[bridge] connect failed", JSON.stringify(m.error)); ws.close(); } + else console.error("[bridge] connected to gateway"); + return; + } + // Streaming deltas: the gateway broadcasts "agent" events to operator + // clients while a run executes. Assistant-stream events carry accumulated + // text and/or the raw delta in payload.data. Events are keyed by the + // gateway-assigned runId, which the ack frame maps to our turnId. + if (m.type === "event" && m.event === "agent") { + const pl = m.payload || {}; + const tid = pl.runId ? runToTurn.get(pl.runId) : null; + const t = tid ? turns.get(tid) : null; + if (t && pl.stream === "assistant" && pl.data) { + if (typeof pl.data.text === "string" && pl.data.text.length >= t.text.length) { + t.text = pl.data.text; + } else if (typeof pl.data.delta === "string") { + t.text += pl.data.delta; + } + } + return; + } + if (m.type === "res" && pending.has(m.id)) { + // The agent method emits TWO res frames with the same id, BOTH ok:true: + // 1) ack: payload.status === "accepted" (+ runId of the spawned run) + // 2) final: payload.status === "ok" + payload.result.payloads[] + // Everything of interest is nested under m.payload (not m.result). + const pl = m.payload || {}; + const result = pl.result; + const isError = m.ok === false || m.error || pl.status === "error"; + const isFinal = (result && Array.isArray(result.payloads)) || isError; + if (!isFinal) { + // ack — capture the gateway runId so delta events route to this turn + if (pl.runId && turns.has(m.id)) runToTurn.set(pl.runId, m.id); + return; + } + const p = pending.get(m.id); + pending.delete(m.id); + clearTimeout(p.timer); + if (isError) { + p.reject(new Error(JSON.stringify(m.error || pl.error || m))); + } else { + p.resolve(result.payloads.map((x) => x.text).filter(Boolean).join(" ") || "(no reply)"); + } + } + }); + ws.on("close", () => { connected = false; setTimeout(connect, 1000); }); + ws.on("error", (e) => { console.error("[bridge] ws error", e.message); }); +} + +function runAgent(message, sessionKey, attachments, turnId, timeoutMs) { + return new Promise((resolve, reject) => { + if (!connected) return reject(new Error("gateway not connected")); + const id = turnId || crypto.randomUUID(); + const timer = setTimeout(() => { + pending.delete(id); reject(new Error("agent turn timeout")); + }, timeoutMs || 280000); + pending.set(id, { resolve, reject, timer }); + // Attachment wire format matches the gateway's own subagent-spawn caller: + // [{type:"image", source:{type:"base64", media_type, data}}] + const gwAttachments = (attachments || []) + .filter((a) => a && a.data && a.media_type) + .map((a) => ({ + type: "image", + source: { type: "base64", media_type: a.media_type, data: a.data }, + })); + ws.send(JSON.stringify({ + type: "req", id, method: "agent", + params: { + message, sessionKey, idempotencyKey: id, + ...(gwAttachments.length ? { attachments: gwAttachments } : {}), + }, + })); + }); +} + +async function readJsonBody(req) { + const chunks = []; + for await (const c of req) chunks.push(c); + return JSON.parse(Buffer.concat(chunks).toString() || "{}"); +} + +http.createServer(async (req, res) => { + const u = new URL(req.url, "http://x"); + if (u.pathname === "/ready") { + res.writeHead(200, { "Content-Type": "application/json" }); + return res.end(JSON.stringify({ connected })); + } + if (u.pathname === "/agent-async" && req.method === "POST") { + // Fire an agent turn and return immediately; poll /progress for text. + let body; + try { body = await readJsonBody(req); } catch { + res.writeHead(400, { "Content-Type": "application/json" }); + return res.end(JSON.stringify({ error: "bad json body" })); + } + gcTurns(); + const turnId = crypto.randomUUID(); + turns.set(turnId, { text: "", done: false, reply: null, error: null, at: Date.now() }); + runAgent(body.m || "", body.s || "default", body.attachments || null, turnId, + ASYNC_TURN_TIMEOUT_MS) + .then((reply) => { + const t = turns.get(turnId); + if (t) { t.done = true; t.reply = reply; t.at = Date.now(); } + }) + .catch((e) => { + console.error("[bridge] async agent turn failed:", e); + const t = turns.get(turnId); + if (t) { t.done = true; t.error = "agent turn failed"; t.at = Date.now(); } + }); + res.writeHead(200, { "Content-Type": "application/json" }); + return res.end(JSON.stringify({ turnId })); + } + if (u.pathname === "/progress") { + const t = turns.get(u.searchParams.get("id") || ""); + if (!t) { + res.writeHead(404, { "Content-Type": "application/json" }); + return res.end(JSON.stringify({ error: "unknown turn" })); + } + t.at = Date.now(); // being polled = alive; see gcTurns + + res.writeHead(200, { "Content-Type": "application/json" }); + return res.end(JSON.stringify({ + text: t.text, done: t.done, + ...(t.reply !== null ? { reply: t.reply } : {}), + ...(t.error ? { error: t.error } : {}), + })); + } + if (u.pathname === "/agent") { + let msg = u.searchParams.get("m") || ""; + let sess = u.searchParams.get("s") || "default"; + let attachments = null; + if (req.method === "POST") { + // JSON body variant for image turns (base64 too big for a query string). + try { + const chunks = []; + for await (const c of req) chunks.push(c); + const body = JSON.parse(Buffer.concat(chunks).toString() || "{}"); + msg = body.m || msg; + sess = body.s || sess; + attachments = body.attachments || null; + } catch { + res.writeHead(400, { "Content-Type": "application/json" }); + return res.end(JSON.stringify({ error: "bad json body" })); + } + } + try { + const reply = await runAgent(msg, sess, attachments); + res.writeHead(200, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ reply })); + } catch (e) { + // Log the detail server-side only; never echo exception text to the client + // (CodeQL js/stack-trace-exposure). + console.error("[bridge] agent turn failed:", e); + res.writeHead(500, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ error: "agent turn failed" })); + } + return; + } + res.writeHead(404); res.end(); +}).listen(PORT, "127.0.0.1", () => console.error(`[bridge] http on :${PORT}`)); + +connect(); diff --git a/lambda-microvms-multi-tenant-ai-agents/src/microvm/hooks.py b/lambda-microvms-multi-tenant-ai-agents/src/microvm/hooks.py new file mode 100644 index 000000000..1f8a719da --- /dev/null +++ b/lambda-microvms-multi-tenant-ai-agents/src/microvm/hooks.py @@ -0,0 +1,152 @@ +"""Sidecar (multi-tenant EFS variant): lifecycle hooks + health + /tenant + /chat + /tg + /files.""" +import json +import os +import subprocess +import urllib.error +import urllib.parse +import urllib.request +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +HOOK = "/aws/lambda-microvms/runtime/v1/" +OPENCLAW = "http://127.0.0.1:18789" +GATEWAY_TOKEN = os.environ.get("OPENCLAW_GATEWAY_TOKEN", "poc-microvm-token-42") +TENANT_FILE = "/var/run/tenant-id" +MARKER = "/var/run/efs-mounted" + + +def agent_turn(message: str, session: str, attachments=None) -> bytes: + """Run one agent turn via the PERSISTENT gateway bridge (no per-message CLI spawn). + + The bridge holds a warm WebSocket to the gateway; a turn is ~2s instead of ~22s. + Image attachments (base64) POST to the bridge; text-only turns keep the GET path. + Returns the same JSON shape callers expect: {"result":{"payloads":[{"text":...}]}}. + """ + if attachments: + req = urllib.request.Request( + "http://127.0.0.1:8090/agent", + data=json.dumps({"m": message, "s": session, + "attachments": attachments}).encode(), + headers={"Content-Type": "application/json"}, method="POST") + with urllib.request.urlopen(req, timeout=290) as r: + d = json.loads(r.read()) + else: + url = ("http://127.0.0.1:8090/agent?m=" + urllib.parse.quote(message) + + "&s=" + urllib.parse.quote(session)) + with urllib.request.urlopen(url, timeout=290) as r: + d = json.loads(r.read()) + if "error" in d: + return json.dumps({"error": d["error"]}).encode() + return json.dumps({"result": {"payloads": [{"text": d.get("reply", "")}]}}).encode() + + +def check(path: str): + try: + with urllib.request.urlopen(OPENCLAW + path, timeout=3) as r: + return r.status, None + except Exception as e: + return getattr(e, "code", None), f"{type(e).__name__}: {e}" + + +def state_report() -> bytes: + r = subprocess.run( + ["sh", "-c", + "echo '--- tenant ---'; cat /var/run/tenant-id 2>&1; " + "echo; echo '--- mounts ---'; grep -E 'efs|openclaw|nfs' /proc/mounts; " + "echo '--- marker ---'; ls -la /var/run/efs-mounted 2>&1; " + "echo '--- state dir ---'; ls -la /home/node/.openclaw/ 2>&1 | head -14; " + "echo '--- sessions ---'; ls -la /home/node/.openclaw/agents/main/sessions/ 2>&1 | head -8; " + "echo '--- efs err ---'; tail -3 /tmp/efs-mount.err 2>&1"], + capture_output=True, text=True, timeout=10, + ) + return (r.stdout + r.stderr).encode() + + +class H(BaseHTTPRequestHandler): + def _send(self, code, body=b""): + self.send_response(code) + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(body) + + def do_GET(self): + if self.path == "/health": + hz, herr = check("/healthz") + rz, rerr = check("/readyz") + self._send(200 if hz == 200 else 503, json.dumps( + {"healthz": hz, "readyz": rz, "efsReady": os.path.exists(MARKER), + "tenant": open(TENANT_FILE).read().strip() if os.path.exists(TENANT_FILE) else None, + "healthzErr": herr, "readyzErr": rerr}).encode()) + elif self.path == "/files": + self._send(200, state_report()) + elif self.path.startswith("/progress"): + # Proxy to the bridge: poll accumulated stream text for an async turn. + try: + with urllib.request.urlopen( + "http://127.0.0.1:8090" + self.path, timeout=10) as r: + self._send(200, r.read()) + except urllib.error.HTTPError as e: + self._send(e.code, e.read()) + except Exception as e: + self._send(500, json.dumps({"error": str(e)}).encode()) + elif self.path.startswith("/chat"): + q = urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query) + msg = (q.get("m") or ["Say pong"])[0] + sess = (q.get("s") or ["poc-demo"])[0] + try: + self._send(200, agent_turn(msg, sess, None)) + except Exception as e: + self._send(500, json.dumps({"error": str(e)}).encode()) + else: + self._send(404) + + def do_POST(self): + if self.path == "/chat": + # JSON body variant of GET /chat — used for turns with image attachments + # (base64 payloads don't fit in a query string). + n = int(self.headers.get("Content-Length") or 0) + try: + d = json.loads(self.rfile.read(n) or b"{}") + msg = d.get("m") or "Say pong" + sess = d.get("s") or "poc-demo" + atts = d.get("attachments") or None + self._send(200, agent_turn(msg, sess, atts)) + except Exception as e: + self._send(500, json.dumps({"error": str(e)}).encode()) + elif self.path == "/chat-async": + # Start a turn without blocking; returns {"turnId"} for /progress polling. + n = int(self.headers.get("Content-Length") or 0) + try: + body = self.rfile.read(n) or b"{}" + json.loads(body) # validate before proxying + req = urllib.request.Request( + "http://127.0.0.1:8090/agent-async", data=body, + headers={"Content-Type": "application/json"}, method="POST") + with urllib.request.urlopen(req, timeout=15) as r: + self._send(200, r.read()) + except Exception as e: + self._send(500, json.dumps({"error": str(e)}).encode()) + elif self.path == "/tenant": + # Orchestrator assigns the tenant; unblocks efs-monitor. + n = int(self.headers.get("Content-Length") or 0) + tid = json.loads(self.rfile.read(n) or b"{}").get("tenantId", "") + tid = "".join(c for c in tid if c.isalnum() or c in "-_")[:64] + if not tid: + self._send(400, b'{"error":"tenantId required"}') + return + with open(TENANT_FILE, "w") as f: + f.write(tid) + print(f"[hooks] tenant assigned: {tid}", flush=True) + self._send(200, json.dumps({"tenant": tid}).encode()) + elif self.path.startswith(HOOK): + print(f"[hooks] POST {self.path} -> 200", flush=True) + self._send(200) + else: + self._send(200) + + def log_message(self, *a): + pass + + +if __name__ == "__main__": + print("[hooks] sidecar listening on :8080", flush=True) + ThreadingHTTPServer(("0.0.0.0", 8080), H).serve_forever() diff --git a/lambda-microvms-multi-tenant-ai-agents/src/microvm/materialize-models.mjs b/lambda-microvms-multi-tenant-ai-agents/src/microvm/materialize-models.mjs new file mode 100644 index 000000000..d89dd23b7 --- /dev/null +++ b/lambda-microvms-multi-tenant-ai-agents/src/microvm/materialize-models.mjs @@ -0,0 +1,59 @@ +// Materialize Bedrock model discovery into openclaw.json at cold start. +// +// Why: the gateway's image-attachment guard and `models list` read only the +// explicit `models.providers..models` config array — they never consult the +// plugin's live discovery catalog (OpenClaw integration gap, verified 2026-07). +// So we run the plugin's own discoverBedrockModels() once per cold start +// (measured ~1.3s warm / ~3.3s cold) and write the result into the config the +// guard does read. Fallback: on any failure the config keeps whatever models +// array it already has (the static seed list baked into the image). +// +// Runs as: node materialize-models.mjs (before gateway starts) +import { readFileSync, writeFileSync, readdirSync } from "node:fs"; + +const CONFIG_PATH = process.argv[2] || "/home/node/.openclaw/openclaw.json"; + +// The plugin project dir carries a content-hash suffix that changes across +// plugin versions — locate it instead of hardcoding. +function findPluginDir() { + const projects = "/home/node/.openclaw/npm/projects"; + const entry = readdirSync(projects).find((d) => + d.startsWith("openclaw-amazon-bedrock-provider-"), + ); + if (!entry) throw new Error("bedrock provider plugin dir not found"); + return `${projects}/${entry}/node_modules/@openclaw/amazon-bedrock-provider`; +} + +try { + const cfg = JSON.parse(readFileSync(CONFIG_PATH, "utf8")); + const provider = cfg?.models?.providers?.["amazon-bedrock"]; + if (!provider) throw new Error("no amazon-bedrock provider in config"); + + const discovery = + cfg?.plugins?.entries?.["amazon-bedrock"]?.config?.discovery ?? {}; + if (discovery.enabled === false) { + console.log("[materialize-models] discovery disabled; keeping static list"); + process.exit(0); + } + + const { discoverBedrockModels } = await import( + `${findPluginDir()}/dist/discovery.js` + ); + const models = await discoverBedrockModels({ + region: discovery.region || process.env.AWS_REGION || "us-east-1", + config: { ...discovery, refreshInterval: 0 }, + }); + if (!Array.isArray(models) || models.length === 0) + throw new Error("discovery returned no models"); + + provider.models = models; + writeFileSync(CONFIG_PATH, JSON.stringify(cfg, null, 1)); + const withImage = models.filter((m) => m.input?.includes("image")).length; + console.log( + `[materialize-models] wrote ${models.length} models (${withImage} vision) to ${CONFIG_PATH}`, + ); +} catch (e) { + console.log( + `[materialize-models] FAILED (${String(e.message).slice(0, 120)}); keeping static list`, + ); +} diff --git a/lambda-microvms-multi-tenant-ai-agents/src/microvm/openclaw.json b/lambda-microvms-multi-tenant-ai-agents/src/microvm/openclaw.json new file mode 100644 index 000000000..f953626ca --- /dev/null +++ b/lambda-microvms-multi-tenant-ai-agents/src/microvm/openclaw.json @@ -0,0 +1,87 @@ +{ + "gateway": { + "mode": "local", + "bind": "custom", + "customBindHost": "0.0.0.0", + "port": 18789, + "auth": { "mode": "token" }, + "controlUi": { "allowedOrigins": ["*"] } + }, + "models": { + "providers": { + "amazon-bedrock": { + "baseUrl": "https://bedrock-runtime.us-east-1.amazonaws.com", + "api": "bedrock-converse-stream", + "auth": "aws-sdk", + "models": [ + { + "id": "us.anthropic.claude-haiku-4-5-20251001-v1:0", + "name": "Claude Haiku 4.5 (Bedrock)", + "input": ["text", "image"], + "cost": { "input": 0, "output": 0, "cacheRead": 0, "cacheWrite": 0 }, + "contextWindow": 200000, + "maxTokens": 8192 + }, + { + "id": "us.anthropic.claude-sonnet-5", + "name": "Claude Sonnet 5 (Bedrock)", + "input": ["text", "image"], + "cost": { "input": 0, "output": 0, "cacheRead": 0, "cacheWrite": 0 }, + "contextWindow": 1000000, + "maxTokens": 16384 + }, + { + "id": "us.anthropic.claude-sonnet-4-6", + "name": "Claude Sonnet 4.6 (Bedrock)", + "input": ["text", "image"], + "cost": { "input": 0, "output": 0, "cacheRead": 0, "cacheWrite": 0 }, + "contextWindow": 1000000, + "maxTokens": 16384 + }, + { + "id": "global.anthropic.claude-opus-4-8", + "name": "Claude Opus 4.8 (Bedrock)", + "input": ["text", "image"], + "cost": { "input": 0, "output": 0, "cacheRead": 0, "cacheWrite": 0 }, + "contextWindow": 1000000, + "maxTokens": 16384 + }, + { + "id": "us.anthropic.claude-opus-4-7", + "name": "Claude Opus 4.7 (Bedrock)", + "input": ["text", "image"], + "cost": { "input": 0, "output": 0, "cacheRead": 0, "cacheWrite": 0 }, + "contextWindow": 1000000, + "maxTokens": 16384 + } + ] + } + } + }, + "agents": { + "defaults": { + "model": { "primary": "amazon-bedrock/us.anthropic.claude-haiku-4-5-20251001-v1:0" }, + "heartbeat": { "every": "0m" } + } + }, + "tools": { + "web": { + "search": { "enabled": true, "provider": "duckduckgo" }, + "fetch": { "enabled": true } + } + }, + "plugins": { + "allow": ["amazon-bedrock"], + "entries": { + "amazon-bedrock": { + "config": { + "discovery": { + "enabled": true, + "region": "us-east-1", + "providerFilter": ["anthropic"] + } + } + } + } + } +} diff --git a/lambda-microvms-multi-tenant-ai-agents/src/microvm/start.sh b/lambda-microvms-multi-tenant-ai-agents/src/microvm/start.sh new file mode 100644 index 000000000..95721e7fd --- /dev/null +++ b/lambda-microvms-multi-tenant-ai-agents/src/microvm/start.sh @@ -0,0 +1,29 @@ +#!/usr/bin/env bash +# MicroVM entrypoint (EFS variant): hook sidecar (8080) + EFS mount daemon + +# supervised OpenClaw gateway (18789). Runs as root so we can mount NFS. +set -u +export OPENCLAW_GATEWAY_TOKEN="${OPENCLAW_GATEWAY_TOKEN:-poc-microvm-token-42}" +export HOME=/home/node + +python3 /opt/poc/hooks.py & +echo "[start] hook sidecar on :8080 (pid $!)" + +/opt/poc/efs-monitor.sh & +echo "[start] efs mount daemon (pid $!)" + +# Persistent gateway bridge (warm WS to :18789) — kills per-message CLI spawn cost. +# Supervise it so it comes back if the gateway bounces during EFS adoption. +( while true; do node /opt/poc/gw-bridge.cjs; echo "[start] gw-bridge exited; restart in 2s"; sleep 2; done ) & +echo "[start] gw-bridge supervisor on :8090 (pid $!)" + +# Supervise the gateway: efs-monitor kills it once EFS is bound so it +# restarts against the EFS-backed state dir. +while true; do + node /app/openclaw.mjs gateway \ + --allow-unconfigured \ + --port 18789 \ + --auth token \ + --token "$OPENCLAW_GATEWAY_TOKEN" + echo "[start] gateway exited (rc=$?); restarting in 2s" + sleep 2 +done diff --git a/lambda-microvms-multi-tenant-ai-agents/src/orchestrator/handler.py b/lambda-microvms-multi-tenant-ai-agents/src/orchestrator/handler.py new file mode 100644 index 000000000..64828e306 --- /dev/null +++ b/lambda-microvms-multi-tenant-ai-agents/src/orchestrator/handler.py @@ -0,0 +1,526 @@ +"""Multi-tenant OpenClaw orchestrator on Lambda MicroVMs — Router + Worker + Sweeper. + +One Lambda, three roles selected by event shape: + - Function URL HTTP event -> ROUTER (fast: ACK Telegram in <1s, hand off to worker) + - {"_worker": {...}} -> WORKER (async self-invoke: ensure VM, run turn, reply; + turns outliving one invocation relay to a successor + via {"_worker": {"resume": ...}} — see stream_turn_to_telegram) + - {"_sweeper": true} (EventBridge) -> SWEEPER (reap idle, reconcile) + +Registry (DynamoDB): tenantId -> microvmId, endpoint, generation, state, launchedAt, +lastActiveAt, botToken(optional), webhookSecret(optional). + +State lifecycle: a tenant's VM is disposable; state lives on per-tenant EFS subdir. +Router NEVER proactively renews (see design-orchestrator.md). Two branches only: +alive -> forward; dead -> cold-start. +""" +import json +import os +import time +import urllib.error +import urllib.parse +import urllib.request + +import boto3 +from boto3.dynamodb.conditions import Attr + +REGION = os.environ["AWS_REGION"] +TABLE = os.environ["TENANTS_TABLE"] +FN_NAME = os.environ["AWS_LAMBDA_FUNCTION_NAME"] +IMAGE_ARN = os.environ["IMAGE_ARN"] +IMAGE_VERSION = os.environ["IMAGE_VERSION"] +EXEC_ROLE_ARN = os.environ["EXEC_ROLE_ARN"] +INGRESS = os.environ["INGRESS_CONNECTOR"] +EGRESS = os.environ["EGRESS_CONNECTOR"] +IDLE_REAP_SECONDS = int(os.environ.get("IDLE_REAP_SECONDS", "3600")) +# A turn that outlives one worker invocation is handed to a fresh one (chained +# self-invoke). 120 hops x ~4.5min (300s timeout) covers the VM's full 8h life. +MAX_TURN_HOPS = int(os.environ.get("MAX_TURN_HOPS", "120")) + +mv = boto3.client("lambda-microvms", region_name=REGION) +lam = boto3.client("lambda", region_name=REGION) +ddb = boto3.resource("dynamodb", region_name=REGION).Table(TABLE) + + +# ---------- helpers ---------- +def now() -> int: + return int(time.time()) + + +def get_tenant(tid): + return ddb.get_item(Key={"tenantId": tid}).get("Item") + + +def mv_state(microvm_id): + try: + return mv.get_microvm(microvmIdentifier=microvm_id).get("state") + except Exception: + return None + + +def call_vm(endpoint, path, token, method="GET", body=None, port=8080, timeout=280): + url = f"https://{endpoint}{path}" + data = json.dumps(body).encode() if body is not None else None + headers = {"X-aws-proxy-auth": token, + "X-aws-proxy-port": str(port), + "Content-Type": "application/json"} + req = urllib.request.Request(url, data=data, method=method, headers=headers) + with urllib.request.urlopen(req, timeout=timeout) as r: + return r.status, r.read() + + +def run_turn(endpoint, token, text, session, attachments=None): + """One agent turn via the sidecar. Images go as a POST body (too big for a URL).""" + if attachments: + st, body = call_vm(endpoint, "/chat", token, "POST", + {"m": text, "s": session, "attachments": attachments}, + timeout=280) + else: + qs = urllib.parse.urlencode({"m": text, "s": session}) + st, body = call_vm(endpoint, f"/chat?{qs}", token, timeout=280) + return st, body + + +def mint_token(microvm_id): + # Only the sidecar (8080) is reachable through the proxy; the OpenClaw gateway + # (18789) stays loopback-only inside the VM. + r = mv.create_microvm_auth_token( + microvmIdentifier=microvm_id, expirationInMinutes=55, + allowedPorts=[{"port": 8080}]) + return r["authToken"]["X-aws-proxy-auth"] + + +def tg_send(bot_token, chat_id, text): + data = urllib.parse.urlencode({"chat_id": chat_id, "text": text[:4000]}).encode() + with urllib.request.urlopen( + f"https://api.telegram.org/bot{bot_token}/sendMessage", data, timeout=20) as r: + body = json.loads(r.read()) + print(f"[worker] tg_send ok={body.get('ok')} " + f"message_id={body.get('result', {}).get('message_id')} chat={chat_id}", flush=True) + return body + + +def tg_typing(bot_token, chat_id): + try: + data = urllib.parse.urlencode({"chat_id": chat_id, "action": "typing"}).encode() + urllib.request.urlopen( + f"https://api.telegram.org/bot{bot_token}/sendChatAction", data, timeout=10) + except Exception: + pass + + +def tg_send_placeholder(bot_token, chat_id): + """Send the placeholder that streaming edits will grow. Returns message_id or None.""" + try: + data = urllib.parse.urlencode({"chat_id": chat_id, "text": "…"}).encode() + with urllib.request.urlopen( + f"https://api.telegram.org/bot{bot_token}/sendMessage", data, timeout=20) as r: + return json.loads(r.read()).get("result", {}).get("message_id") + except Exception as e: + print(f"[worker] placeholder send failed: {e}", flush=True) + return None + + +def tg_edit(bot_token, chat_id, message_id, text): + """Edit a message in place. Telegram rejects edits with unchanged text — skip those.""" + try: + data = urllib.parse.urlencode({ + "chat_id": chat_id, "message_id": message_id, "text": text[:4000]}).encode() + urllib.request.urlopen( + f"https://api.telegram.org/bot{bot_token}/editMessageText", data, timeout=20) + return True + except urllib.error.HTTPError as e: + print(f"[worker] edit failed: {e} body={e.read()[:200]}", flush=True) + return False + except Exception as e: + print(f"[worker] edit failed: {e}", flush=True) + return False + + +def stream_turn_to_telegram(endpoint, token, text, session, attachments, + bot, chat_id, tid, ctx, resume=None): + """Pseudo-streaming: async turn + poll /progress, growing one Telegram message. + + Telegram has no true streaming — the standard technique is a placeholder + message repeatedly edited via editMessageText (~1 edit/sec/chat allowed; + we edit every ~2s). Falls back to the sync path if the async start fails. + + A turn that outlives this Lambda invocation is NOT lost: just before the + invocation times out, polling is handed to a fresh async self-invoke + (carrying turnId + message state), the same chaining the router uses for + the webhook handoff. Returns None on handoff — the successor owns delivery. + A single turn is thus bounded by the VM's 8h lifetime, not Lambda's 15 min. + """ + if resume: + turn_id = resume["turnId"] + msg_id = resume.get("msgId") + last_len = int(resume.get("lastLen", 0)) + hops = int(resume.get("hops", 0)) + else: + st, body = call_vm(endpoint, "/chat-async", token, "POST", + {"m": text, "s": session, "attachments": attachments or []}, + timeout=30) + turn_id = json.loads(body).get("turnId") + if not turn_id: + raise RuntimeError("no turnId from /chat-async") + msg_id = tg_send_placeholder(bot, chat_id) + last_len, hops = 0, 0 + + reply = None + # 1.2s ≈ Telegram's per-chat edit ceiling (~1/s) with headroom; a failed + # edit (e.g. 429) leaves last_len unchanged so the next cycle retries. + while True: + # Hand off before this invocation's timeout would silently kill polling. + # 20s leaves room for one in-flight poll + the invoke round-trip. + if ctx.get_remaining_time_in_millis() < 20_000: + if hops + 1 >= MAX_TURN_HOPS: + reply = "(turn exceeded the relay budget — increase MAX_TURN_HOPS)" + break + lam.invoke(FunctionName=FN_NAME, InvocationType="Event", + Payload=json.dumps({"_worker": {"tenantId": tid, "resume": { + "turnId": turn_id, "chatId": chat_id, "msgId": msg_id, + "lastLen": last_len, "hops": hops + 1}}}).encode()) + print(f"[worker] turn {turn_id} handed off (hop {hops + 1})", flush=True) + return None + time.sleep(1.2) + try: + st, body = call_vm(endpoint, f"/progress?id={turn_id}", token, timeout=15) + prog = json.loads(body) + except urllib.error.HTTPError as e: + if e.code == 404: + # Bridge no longer knows the turn (VM replaced / bridge restarted + # mid-turn) — retrying or relaying can never recover it. + reply = "(turn lost — the VM was replaced mid-turn, please retry)" + break + continue + except Exception: + continue + if prog.get("done"): + reply = prog.get("reply") or prog.get("text") or "(no reply)" + if prog.get("error"): + reply = "(agent error — try again)" + break + cur = prog.get("text") or "" + # Grow the placeholder with partial text; "▌" cursor marks in-progress. + # Stop editing once past Telegram's 4096-char message cap: tg_edit + # truncates, so further edits would be byte-identical -> 400 spam. + if msg_id and len(cur) > last_len and last_len < 3990: + if tg_edit(bot, chat_id, msg_id, cur + " ▌"): + last_len = len(cur) + if msg_id: + tg_edit(bot, chat_id, msg_id, reply) + else: + tg_send(bot, chat_id, reply) + return reply + + +def tg_fetch_image(bot_token, file_id, max_bytes=4_500_000): + """Download a Telegram file and return (base64, media_type), or (None, None). + + Caps at ~4.5MB — Bedrock's per-image limit is 5MB and Telegram photos + (recompressed JPEG) are far smaller; the cap only bites on image documents. + """ + import base64 + import mimetypes + try: + with urllib.request.urlopen( + f"https://api.telegram.org/bot{bot_token}/getFile?file_id=" + + urllib.parse.quote(file_id), timeout=20) as r: + path = json.loads(r.read())["result"]["file_path"] + with urllib.request.urlopen( + f"https://api.telegram.org/file/bot{bot_token}/{path}", timeout=60) as r: + data = r.read(max_bytes + 1) + if len(data) > max_bytes: + print(f"[worker] image too large ({len(data)}B), skipped", flush=True) + return None, None + media_type = mimetypes.guess_type(path)[0] or "image/jpeg" + return base64.b64encode(data).decode(), media_type + except Exception as e: + print(f"[worker] tg_fetch_image failed: {e}", flush=True) + return None, None + + +def extract_content(msg, bot): + """Pull text + image attachments out of a Telegram message. + + Photos live in msg.photo[] (sizes ascending; last is largest) with the + user's text in msg.caption — there is NO msg.text on media messages. + Image documents (sent as files) come via msg.document with an image/* mime. + """ + text = msg.get("text", "") or msg.get("caption", "") + attachments = [] + if bot: + photos = msg.get("photo") or [] + if photos: + b64, mt = tg_fetch_image(bot, photos[-1]["file_id"]) + if b64: + attachments.append({"media_type": mt, "data": b64}) + doc = msg.get("document") or {} + if (doc.get("mime_type") or "").startswith("image/"): + b64, _ = tg_fetch_image(bot, doc["file_id"]) + if b64: + attachments.append({"media_type": doc["mime_type"], "data": b64}) + if attachments and not text: + text = "(the user sent this image with no caption — look at it and respond)" + return text, attachments + + +# ---------- cold start (single-writer lock via conditional update) ---------- +def cold_start(tid, item): + gen = int(item.get("generation", 0)) + 1 + # Lock: only one launcher may flip state PENDING for this generation. + try: + ddb.update_item( + Key={"tenantId": tid}, + UpdateExpression="SET #s=:p, generation=:g, launchedAt=:t", + ConditionExpression=Attr("state").ne("LAUNCHING"), + ExpressionAttributeNames={"#s": "state"}, + ExpressionAttributeValues={":p": "LAUNCHING", ":g": gen, ":t": now()}) + except Exception: + # Another invocation is already launching; wait for it to publish endpoint. + for _ in range(40): + time.sleep(3) + it = get_tenant(tid) + if it and it.get("state") == "RUNNING" and it.get("endpoint"): + return it + raise RuntimeError("concurrent launch did not converge") + + r = mv.run_microvm( + imageIdentifier=IMAGE_ARN, imageVersion=IMAGE_VERSION, + executionRoleArn=EXEC_ROLE_ARN, + ingressNetworkConnectors=[INGRESS], + egressNetworkConnectors=[EGRESS], + idlePolicy={"autoResumeEnabled": True, "maxIdleDurationSeconds": 900, + "suspendedDurationSeconds": 3600}, + maximumDurationInSeconds=28800) + microvm_id, endpoint = r["microvmId"], r["endpoint"] + + # wait RUNNING + for _ in range(40): + if mv_state(microvm_id) == "RUNNING": + break + time.sleep(3) + token = mint_token(microvm_id) + + # assign tenant -> unblocks efs-monitor -> mounts per-tenant subdir -> bounces gateway + for _ in range(20): + try: + call_vm(endpoint, "/tenant", token, "POST", {"tenantId": tid}, timeout=15) + break + except Exception: + time.sleep(3) + + # gate on EFS adoption + gateway healthy + ready = False + for _ in range(40): + try: + st, body = call_vm(endpoint, "/health", token, timeout=15) + h = json.loads(body) + if h.get("efsReady") and h.get("healthz") == 200: + ready = True + break + except Exception: + pass + time.sleep(3) + + item = {**item, "tenantId": tid, "microvmId": microvm_id, "endpoint": endpoint, + "generation": gen, "state": "RUNNING", "launchedAt": now(), + "lastActiveAt": now(), "authToken": token} + ddb.put_item(Item=item) + if not ready: + raise RuntimeError("cold start: VM did not become EFS-ready") + return item + + +def ensure_vm(tid, item): + """Two-branch decision: alive -> reuse; dead -> cold start.""" + mid = item.get("microvmId") + state = mv_state(mid) if mid else None + if state in ("RUNNING", "SUSPENDED"): + # alive (SUSPENDED auto-resumes on the forwarded request) + return item, False + return cold_start(tid, item), True + + +# ---------- WORKER (async) ---------- +def resume_turn(payload, ctx): + """Successor invocation in a turn relay: keep polling an in-flight turn.""" + tid = payload["tenantId"] + resume = payload["resume"] + item = get_tenant(tid) + bot, chat_id = item.get("botToken"), resume["chatId"] + try: + endpoint = item["endpoint"] + token = mint_token(item["microvmId"]) + except Exception as e: + # VM died mid-turn (e.g. hit its 8h max lifetime) — tell the user + # instead of failing silently. + print(f"[worker] resume for {tid} found no live VM: {e}", flush=True) + note = "(turn lost — the VM ended mid-turn, please retry)" + if resume.get("msgId"): + tg_edit(bot, chat_id, resume["msgId"], note) + else: + tg_send(bot, chat_id, note) + return {"tenant": tid, "resumed": False, "error": "vm gone"} + reply = stream_turn_to_telegram(endpoint, token, None, None, None, + bot, chat_id, tid, ctx, resume=resume) + if reply is None: + return {"tenant": tid, "handedOff": True} + ddb.update_item(Key={"tenantId": tid}, + UpdateExpression="SET lastActiveAt=:t", + ExpressionAttributeValues={":t": now()}) + return {"tenant": tid, "resumed": True, "reply": reply[:80]} + + +def worker(payload, ctx): + tid = payload["tenantId"] + update = payload["update"] + item = get_tenant(tid) + bot = item.get("botToken") + msg = update.get("message") or {} + chat_id = str((msg.get("chat") or {}).get("id", "")) + text, attachments = extract_content(msg, bot) + if not text or not chat_id: + return {"skipped": "no content/chat"} + + # Telegram chat ids are numeric; chat.sh passes a session string (e.g. "cli") + # here. Only numeric ids go down the Telegram send/edit path — otherwise a + # Telegram-enabled tenant crashes every chat.sh turn on sendMessage 400. + is_tg = bool(bot) and chat_id.lstrip("-").isdigit() + if is_tg: + tg_typing(bot, chat_id) + + item, cold = ensure_vm(tid, item) + endpoint = item["endpoint"] + # Always mint a fresh token per turn: token TTL (<=60min) < VM lifetime (8h), + # so a cached token from cold-start time is often already expired -> 403. + token = mint_token(item["microvmId"]) + + session = f"tg-{chat_id}" + if is_tg: + # Telegram path: pseudo-stream via placeholder + editMessageText. + try: + reply = stream_turn_to_telegram( + endpoint, token, text, session, attachments, bot, chat_id, + tid, ctx) + if reply is None: + # Turn outlived this invocation; a chained self-invoke now owns + # polling + delivery. lastActiveAt is updated by the last hop. + return {"tenant": tid, "cold": cold, "handedOff": True} + except Exception as e: + # Streaming plumbing failed (old image, bridge restart, ...): + # fall back to the sync turn + single send. + print(f"[worker] stream path failed ({e}); falling back to sync", flush=True) + st, body = run_turn(endpoint, token, text, session, attachments) + d = json.loads(body) + reply = " ".join(p.get("text", "") for p in + d.get("result", {}).get("payloads", [])) or "(no reply)" + tg_send(bot, chat_id, reply) + else: + # No bot token (chat.sh / HTTP test path): synchronous turn, unchanged. + st, body = run_turn(endpoint, token, text, session, attachments) + d = json.loads(body) + reply = " ".join(p.get("text", "") for p in + d.get("result", {}).get("payloads", [])) or "(no reply)" + ddb.update_item(Key={"tenantId": tid}, + UpdateExpression="SET lastActiveAt=:t", + ExpressionAttributeValues={":t": now()}) + return {"tenant": tid, "cold": cold, "reply": reply[:80]} + + +# ---------- ROUTER (Function URL) ---------- +def router(event): + raw = event.get("rawPath", "/") + body = event.get("body") or "" + if event.get("isBase64Encoded"): + import base64 + body = base64.b64decode(body).decode() + headers = {k.lower(): v for k, v in (event.get("headers") or {}).items()} + + # /tg/ + parts = [p for p in raw.split("/") if p] + if len(parts) == 2 and parts[0] == "tg": + tid = parts[1] + item = get_tenant(tid) + if not item: + return {"statusCode": 404, "body": "unknown tenant"} + # verify Telegram secret token + want = item.get("webhookSecret") + got = headers.get("x-telegram-bot-api-secret-token") + if want and got != want: + return {"statusCode": 403, "body": "bad secret"} + update = json.loads(body or "{}") + # hand off to worker asynchronously; ACK Telegram immediately + lam.invoke(FunctionName=FN_NAME, InvocationType="Event", + Payload=json.dumps({"_worker": {"tenantId": tid, "update": update}}).encode()) + return {"statusCode": 200, "body": "ok"} + + # /chat/?m=... — synchronous test entry (no Telegram); ensures VM + runs a turn inline. + if len(parts) == 2 and parts[0] == "chat": + tid = parts[1] + item = get_tenant(tid) + if not item: + return {"statusCode": 404, "body": "unknown tenant"} + qs = urllib.parse.parse_qs(event.get("rawQueryString", "")) + msg = (qs.get("m") or ["Say pong"])[0] + sess = (qs.get("s") or ["http-demo"])[0] + try: + item, cold = ensure_vm(tid, item) + token = mint_token(item["microvmId"]) # fresh per call (see worker note) + q2 = urllib.parse.urlencode({"m": msg, "s": sess}) + st, body = call_vm(item["endpoint"], f"/chat?{q2}", token, timeout=280) + ddb.update_item(Key={"tenantId": tid}, + UpdateExpression="SET lastActiveAt=:t", + ExpressionAttributeValues={":t": now()}) + d = json.loads(body) + reply = " ".join(p.get("text", "") for p in + d.get("result", {}).get("payloads", [])) or body.decode()[:200] + return {"statusCode": 200, "body": json.dumps({"tenant": tid, "cold": cold, "reply": reply})} + except Exception as e: + return {"statusCode": 500, "body": json.dumps({"error": str(e)})} + + if raw == "/health": + return {"statusCode": 200, "body": json.dumps({"ok": True, "role": "router"})} + return {"statusCode": 404, "body": "not found"} + + +# ---------- SWEEPER (EventBridge) ---------- +def sweeper(): + reaped, reconciled = [], [] + for item in ddb.scan().get("Items", []): + tid, mid = item["tenantId"], item.get("microvmId") + if not mid: + continue + state = mv_state(mid) + if state is None or state == "TERMINATED": + if item.get("state") != "COLD": + ddb.update_item(Key={"tenantId": tid}, + UpdateExpression="SET #s=:c", + ExpressionAttributeNames={"#s": "state"}, + ExpressionAttributeValues={":c": "COLD"}) + reconciled.append(tid) + continue + idle = now() - int(item.get("lastActiveAt", 0)) + if state == "SUSPENDED" and idle > IDLE_REAP_SECONDS: + try: + mv.terminate_microvm(microvmIdentifier=mid) + ddb.update_item(Key={"tenantId": tid}, + UpdateExpression="SET #s=:c", + ExpressionAttributeNames={"#s": "state"}, + ExpressionAttributeValues={":c": "COLD"}) + reaped.append(tid) + except Exception: + pass + return {"reaped": reaped, "reconciled": reconciled} + + +def handler(event, context): + if isinstance(event, dict) and "_worker" in event: + if "resume" in event["_worker"]: + return resume_turn(event["_worker"], context) + return worker(event["_worker"], context) + if isinstance(event, dict) and event.get("_sweeper"): + return sweeper() + if isinstance(event, dict) and ("rawPath" in event or "requestContext" in event): + return router(event) + return {"statusCode": 400, "body": "unrecognized event"} diff --git a/lambda-microvms-multi-tenant-ai-agents/teardown.sh b/lambda-microvms-multi-tenant-ai-agents/teardown.sh new file mode 100755 index 000000000..264f2a1f0 --- /dev/null +++ b/lambda-microvms-multi-tenant-ai-agents/teardown.sh @@ -0,0 +1,49 @@ +#!/usr/bin/env bash +# Tear down everything. CloudFormation now owns the MicroVM image and the VPC egress +# connector, so it removes those itself — we only have to (a) terminate running MicroVMs +# first (they're created imperatively by the orchestrator at runtime, NOT stack-managed, +# and their ENIs/connector use would block stack deletion), then (b) delete the stack, +# then (c) empty & drop the artifact bucket (also not stack-managed). +# Usage: ./teardown.sh STACK REGION +set -uo pipefail +if [ $# -lt 2 ]; then + echo "usage: $0 STACK REGION" >&2; exit 1 +fi +STACK="$1"; REGION="$2" +ACCOUNT="$(aws sts get-caller-identity --query Account --output text)" +BUCKET="${STACK}-artifact-${ACCOUNT}-${REGION}" +say(){ printf '\n== %s ==\n' "$*"; } + +say "1. terminate THIS stack's running MicroVMs (runtime-created, not stack-managed)" +# MicroVM instances carry no stack tag, but get-microvm returns imageArn, and this stack's +# image is named "-openclaw" (see MicrovmImage.Name in template.yaml). Filter on that +# so we never terminate VMs belonging to another stack sharing the account/region. +IMAGE_SUFFIX=":microvm-image:${STACK}-openclaw" +found=0 +for id in $(aws lambda-microvms list-microvms --region "$REGION" \ + --query "items[?state!='TERMINATED'].microvmId" --output text 2>/dev/null); do + arn="$(aws lambda-microvms get-microvm --region "$REGION" --microvm-identifier "$id" \ + --query imageArn --output text 2>/dev/null)" + case "$arn" in + *"$IMAGE_SUFFIX") echo " terminate $id (image: ${arn##*:})"; found=1 + aws lambda-microvms terminate-microvm --region "$REGION" --microvm-identifier "$id" >/dev/null 2>&1 || true ;; + *) echo " skip $id (image: ${arn##*:} — not this stack)" ;; + esac +done +# let ENIs from the connector detach before CFN tries to delete the SG/subnet +[ "$found" = 1 ] && sleep 20 || true + +say "2. delete CloudFormation stack (removes image, connector, EFS, VPC, IAM, DDB, Lambda, API)" +aws cloudformation delete-stack --region "$REGION" --stack-name "$STACK" +echo " waiting for delete to complete..." +if aws cloudformation wait stack-delete-complete --region "$REGION" --stack-name "$STACK" 2>/dev/null; then + echo " stack deleted" +else + echo " NOTE: if delete stalled on the VPC/SG, a MicroVM or the connector's ENIs may still" + echo " be detaching. Re-run this script; CloudFormation delete is idempotent." +fi + +say "3. empty & delete artifact bucket" +aws s3 rm "s3://$BUCKET" --recursive >/dev/null 2>&1 || true +aws s3api delete-bucket --bucket "$BUCKET" --region "$REGION" >/dev/null 2>&1 || true +echo "DONE" diff --git a/lambda-microvms-multi-tenant-ai-agents/template.yaml b/lambda-microvms-multi-tenant-ai-agents/template.yaml new file mode 100644 index 000000000..3d481aead --- /dev/null +++ b/lambda-microvms-multi-tenant-ai-agents/template.yaml @@ -0,0 +1,489 @@ +AWSTemplateFormatVersion: '2010-09-09' +Description: > + Multi-tenant OpenClaw on AWS Lambda MicroVMs — fully declarative infrastructure. + One MicroVM per tenant, per-tenant state on EFS, Bedrock via VPC endpoint, + push (Telegram webhook) orchestrator behind API Gateway. + The MicroVM image and the VPC egress connector are NATIVE CloudFormation resources + (AWS::Lambda::MicrovmImage / AWS::Lambda::NetworkConnector, GA 2026-06-22), so the + only imperative step deploy.sh performs is uploading the two zip artifacts to S3 + (CloudFormation cannot author S3 object *contents*). Everything else is declared here, + and the orchestrator Lambda is wired via GetAtt — no post-deploy CLI patching. + +Parameters: + ProjectName: + Type: String + Default: openclaw-mt + Description: Prefix for named resources. + GatewayToken: + Type: String + NoEcho: true + Default: poc-microvm-token-42 + Description: Shared token OpenClaw's gateway requires (injected into each MicroVM). + BedrockModelId: + Type: String + Default: us.anthropic.claude-haiku-4-5-20251001-v1:0 + Description: Bedrock inference-profile model id used by the agent. + IdleReapSeconds: + Type: Number + Default: 3600 + Description: Sweeper terminates a SUSPENDED tenant VM idle longer than this. + MicrovmImageKey: + Type: String + Default: microvm-images/openclaw.zip + Description: S3 key (in the artifact bucket) of the MicroVM image zip. deploy.sh uploads it before create. + OrchestratorCodeKey: + Type: String + Default: lambda/orchestrator.zip + Description: S3 key of the bundled orchestrator Lambda zip. deploy.sh uploads it before create. + BaseMemoryMiB: + Type: Number + Default: 2048 + Description: MicroVM baseline memory (vCPU scales at 2GB=1vCPU). + ArtifactBucketName: + Type: String + Description: > + Existing S3 bucket holding the uploaded zips. deploy.sh creates it (CloudFormation + cannot pre-populate object contents, and the MicrovmImage/Lambda resources need the + objects to exist at create time), then passes the name here. + +Resources: + + # ---------- Networking: private VM subnet + NAT internet egress ---------- + # The agent needs general internet access (web fetch/search); Bedrock and EFS + # traffic still takes the VPCE/mount-target private paths, bypassing the NAT. + Vpc: + Type: AWS::EC2::VPC + Properties: + CidrBlock: 10.20.0.0/16 + EnableDnsSupport: true # required for VPCE private DNS + EnableDnsHostnames: true + Tags: [{ Key: Name, Value: !Sub '${ProjectName}-vpc' }] + + Subnet: + Type: AWS::EC2::Subnet + Properties: + VpcId: !Ref Vpc + CidrBlock: 10.20.1.0/24 + AvailabilityZone: !Select [0, !GetAZs ''] + Tags: [{ Key: Name, Value: !Sub '${ProjectName}-subnet' }] + + Igw: + Type: AWS::EC2::InternetGateway + Properties: + Tags: [{ Key: Name, Value: !Sub '${ProjectName}-igw' }] + + IgwAttachment: + Type: AWS::EC2::VPCGatewayAttachment + Properties: + VpcId: !Ref Vpc + InternetGatewayId: !Ref Igw + + # Public subnet exists only to host the NAT gateway. + PublicSubnet: + Type: AWS::EC2::Subnet + Properties: + VpcId: !Ref Vpc + CidrBlock: 10.20.0.0/24 + AvailabilityZone: !Select [0, !GetAZs ''] + Tags: [{ Key: Name, Value: !Sub '${ProjectName}-public' }] + + PublicRouteTable: + Type: AWS::EC2::RouteTable + Properties: + VpcId: !Ref Vpc + Tags: [{ Key: Name, Value: !Sub '${ProjectName}-public-rt' }] + + PublicDefaultRoute: + Type: AWS::EC2::Route + DependsOn: IgwAttachment + Properties: + RouteTableId: !Ref PublicRouteTable + DestinationCidrBlock: 0.0.0.0/0 + GatewayId: !Ref Igw + + PublicSubnetRtAssoc: + Type: AWS::EC2::SubnetRouteTableAssociation + Properties: + SubnetId: !Ref PublicSubnet + RouteTableId: !Ref PublicRouteTable + + NatEip: + Type: AWS::EC2::EIP + DependsOn: IgwAttachment + Properties: + Domain: vpc + + NatGateway: + Type: AWS::EC2::NatGateway + Properties: + AllocationId: !GetAtt NatEip.AllocationId + SubnetId: !Ref PublicSubnet + Tags: [{ Key: Name, Value: !Sub '${ProjectName}-nat' }] + + PrivateRouteTable: + Type: AWS::EC2::RouteTable + Properties: + VpcId: !Ref Vpc + Tags: [{ Key: Name, Value: !Sub '${ProjectName}-private-rt' }] + + PrivateDefaultRoute: + Type: AWS::EC2::Route + Properties: + RouteTableId: !Ref PrivateRouteTable + DestinationCidrBlock: 0.0.0.0/0 + NatGatewayId: !Ref NatGateway + + PrivateSubnetRtAssoc: + Type: AWS::EC2::SubnetRouteTableAssociation + Properties: + SubnetId: !Ref Subnet + RouteTableId: !Ref PrivateRouteTable + + # SG shared by EFS mount target, the VM's egress connector ENIs, and the Bedrock VPCE. + # Self-referencing ingress on 2049 (NFS) + 443 (HTTPS to Bedrock). + VpcSg: + Type: AWS::EC2::SecurityGroup + Properties: + GroupDescription: !Sub '${ProjectName} MicroVM egress + EFS + Bedrock VPCE' + VpcId: !Ref Vpc + Tags: [{ Key: Name, Value: !Sub '${ProjectName}-sg' }] + + SgSelfNfs: + Type: AWS::EC2::SecurityGroupIngress + Properties: + GroupId: !Ref VpcSg + IpProtocol: tcp + FromPort: 2049 + ToPort: 2049 + SourceSecurityGroupId: !Ref VpcSg + + SgSelfHttps: + Type: AWS::EC2::SecurityGroupIngress + Properties: + GroupId: !Ref VpcSg + IpProtocol: tcp + FromPort: 443 + ToPort: 443 + SourceSecurityGroupId: !Ref VpcSg + + # ---------- Bedrock runtime VPC endpoint (VM egress goes via VPC, no internet) ---------- + BedrockVpce: + Type: AWS::EC2::VPCEndpoint + Properties: + VpcId: !Ref Vpc + ServiceName: !Sub 'com.amazonaws.${AWS::Region}.bedrock-runtime' + VpcEndpointType: Interface + PrivateDnsEnabled: true + SubnetIds: [!Ref Subnet] + SecurityGroupIds: [!Ref VpcSg] + + # Bedrock control-plane endpoint: the provider plugin's model discovery calls + # ListFoundationModels/ListInferenceProfiles on bedrock. (not -runtime). + BedrockControlVpce: + Type: AWS::EC2::VPCEndpoint + Properties: + VpcId: !Ref Vpc + ServiceName: !Sub 'com.amazonaws.${AWS::Region}.bedrock' + VpcEndpointType: Interface + PrivateDnsEnabled: true + SubnetIds: [!Ref Subnet] + SecurityGroupIds: [!Ref VpcSg] + + # ---------- EFS: per-tenant state lives under /tenants/ on this filesystem ---------- + Efs: + Type: AWS::EFS::FileSystem + Properties: + Encrypted: true + PerformanceMode: generalPurpose + ThroughputMode: elastic + FileSystemTags: [{ Key: Name, Value: !Sub '${ProjectName}-state' }] + + EfsMountTarget: + Type: AWS::EFS::MountTarget + Properties: + FileSystemId: !Ref Efs + SubnetId: !Ref Subnet + SecurityGroups: [!Ref VpcSg] + + # ---------- IAM: MicroVM build role (image creation: read artifact, write build logs) ---------- + MicroVMBuildRole: + Type: AWS::IAM::Role + Properties: + RoleName: !Sub '${ProjectName}-MicroVMBuildRole' + AssumeRolePolicyDocument: + Version: '2012-10-17' + Statement: + - Effect: Allow + Principal: { Service: lambda.amazonaws.com } + Action: [sts:AssumeRole, sts:TagSession] + Condition: { StringEquals: { aws:SourceAccount: !Ref AWS::AccountId } } + Policies: + - PolicyName: build + PolicyDocument: + Version: '2012-10-17' + Statement: + - Effect: Allow + Action: [s3:GetObject] + Resource: !Sub 'arn:aws:s3:::${ArtifactBucketName}/*' + - Effect: Allow + Action: [logs:CreateLogGroup, logs:CreateLogStream, logs:PutLogEvents] + Resource: !Sub 'arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws/lambda-microvms/*' + + # ---------- IAM: MicroVM execution role (runtime: Bedrock via inference profile + logs) ---------- + MicroVMExecutionRole: + Type: AWS::IAM::Role + Properties: + RoleName: !Sub '${ProjectName}-MicroVMExecutionRole' + AssumeRolePolicyDocument: + Version: '2012-10-17' + Statement: + - Effect: Allow + Principal: { Service: lambda.amazonaws.com } + Action: [sts:AssumeRole, sts:TagSession] + Condition: { StringEquals: { aws:SourceAccount: !Ref AWS::AccountId } } + Policies: + - PolicyName: runtime + PolicyDocument: + Version: '2012-10-17' + Statement: + # Not tied to a specific model: OpenClaw may be pointed at different Bedrock + # models (see BedrockModelId, which only sets runtime config). Scope stays on the + # two invoke actions, but any foundation model / inference profile is allowed so + # swapping models needs no IAM change. Cross-region inference profiles need both + # the profile ARN and the underlying foundation-model ARNs in each routed region — + # the '*' region wildcards below cover that for any model. + - Sid: InvokeBedrock + Effect: Allow + Action: [bedrock:InvokeModel, bedrock:InvokeModelWithResponseStream] + Resource: + - !Sub 'arn:aws:bedrock:*::foundation-model/*' + - !Sub 'arn:aws:bedrock:*:${AWS::AccountId}:inference-profile/*' + - !Sub 'arn:aws:bedrock:*:${AWS::AccountId}:application-inference-profile/*' + - Sid: ListForDiscovery + Effect: Allow + Action: [bedrock:ListFoundationModels, bedrock:ListInferenceProfiles] + Resource: '*' + - Sid: RuntimeLogs + Effect: Allow + Action: [logs:CreateLogGroup, logs:CreateLogStream, logs:PutLogEvents] + Resource: !Sub 'arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws/lambda-microvms/*' + + # ---------- IAM: network connector operator role (Lambda-core creates the VPC egress ENIs) ---------- + # NOTE: trust MUST be plain lambda.amazonaws.com WITHOUT aws:SourceAccount — the connector + # service fails to assume a conditioned role ("unable to assume"). Verified live. + NetworkConnectorOperatorRole: + Type: AWS::IAM::Role + Properties: + RoleName: !Sub '${ProjectName}-NetworkConnectorOperatorRole' + AssumeRolePolicyDocument: + Version: '2012-10-17' + Statement: + - Effect: Allow + Principal: { Service: lambda.amazonaws.com } + Action: [sts:AssumeRole, sts:TagSession] + Policies: + - PolicyName: eni + PolicyDocument: + Version: '2012-10-17' + Statement: + - Effect: Allow + Action: + - ec2:CreateNetworkInterface + - ec2:CreateTags + - ec2:DescribeNetworkInterfaces + - ec2:DeleteNetworkInterface + - ec2:DescribeSubnets + - ec2:DescribeSecurityGroups + - ec2:DescribeVpcs + Resource: '*' + + # ---------- MicroVM image (native CFN resource; CFN runs the build on create/update) ---------- + # CodeArtifact points at the zip deploy.sh uploaded to the artifact bucket. All props are + # required by the resource schema. EnvironmentVariables is a Key/Value array. EFS_ID is baked + # so efs-monitor.sh can derive the mount-target DNS name at runtime; AWS_REGION is reserved. + MicrovmImage: + Type: AWS::Lambda::MicrovmImage + Properties: + Name: !Sub '${ProjectName}-openclaw' + BaseImageArn: !Sub 'arn:aws:lambda:${AWS::Region}:aws:microvm-image:al2023-1' + BaseImageVersion: '0' + BuildRoleArn: !GetAtt MicroVMBuildRole.Arn + Description: OpenClaw per-tenant agent image + CodeArtifact: + Uri: !Sub 's3://${ArtifactBucketName}/${MicrovmImageKey}' + CpuConfigurations: + - Architecture: ARM_64 + Resources: + - MinimumMemoryInMiB: !Ref BaseMemoryMiB + AdditionalOsCapabilities: [ALL] + EgressNetworkConnectors: + - !Sub 'arn:aws:lambda:${AWS::Region}:aws:network-connector:aws-network-connector:INTERNET_EGRESS' + Logging: + CloudWatch: + LogGroup: !Sub '/aws/lambda-microvms/${ProjectName}-openclaw' + # No hooks: the app has no /ready hook, and specifying a hooks Port without any hook + # enabled is rejected. The gateway boots fast enough that the platform snapshots after + # ENTRYPOINT init without a readiness gate (verified: hookless builds succeed). + Hooks: + MicrovmHooks: {} + MicrovmImageHooks: {} + EnvironmentVariables: + - Key: OPENCLAW_GATEWAY_TOKEN + Value: !Ref GatewayToken + - Key: EFS_ID + Value: !Ref Efs + + # ---------- VPC egress network connector (native CFN resource) ---------- + EgressConnector: + Type: AWS::Lambda::NetworkConnector + Properties: + Name: !Sub '${ProjectName}-egress' + OperatorRole: !GetAtt NetworkConnectorOperatorRole.Arn + Configuration: + VpcEgressConfiguration: + SubnetIds: [!Ref Subnet] + SecurityGroupIds: [!Ref VpcSg] + NetworkProtocol: IPv4 + AssociatedComputeResourceTypes: [MicroVm] + + # ---------- DynamoDB: tenant registry ---------- + TenantsTable: + Type: AWS::DynamoDB::Table + Properties: + TableName: !Sub '${ProjectName}-tenants' + BillingMode: PAY_PER_REQUEST + AttributeDefinitions: [{ AttributeName: tenantId, AttributeType: S }] + KeySchema: [{ AttributeName: tenantId, KeyType: HASH }] + + # ---------- IAM: orchestrator Lambda role ---------- + OrchestratorRole: + Type: AWS::IAM::Role + Properties: + RoleName: !Sub '${ProjectName}-OrchestratorRole' + AssumeRolePolicyDocument: + Version: '2012-10-17' + Statement: + - Effect: Allow + Principal: { Service: lambda.amazonaws.com } + Action: sts:AssumeRole + Policies: + - PolicyName: orchestrate + PolicyDocument: + Version: '2012-10-17' + Statement: + - Sid: Logs + Effect: Allow + Action: [logs:CreateLogGroup, logs:CreateLogStream, logs:PutLogEvents] + Resource: !Sub 'arn:aws:logs:${AWS::Region}:${AWS::AccountId}:*' + - Sid: MicroVMs + Effect: Allow + Action: + - lambda:RunMicrovm + - lambda:GetMicrovm + - lambda:ListMicrovms + - lambda:SuspendMicrovm + - lambda:ResumeMicrovm + - lambda:TerminateMicrovm + - lambda:CreateMicrovmAuthToken + Resource: '*' + - Sid: PassExecRole + Effect: Allow + Action: iam:PassRole + Resource: !GetAtt MicroVMExecutionRole.Arn + - Sid: PassConnectors + Effect: Allow + Action: lambda:PassNetworkConnector + Resource: + - !Sub 'arn:aws:lambda:${AWS::Region}:aws:network-connector:aws-network-connector:*' + - !Sub 'arn:aws:lambda:${AWS::Region}:${AWS::AccountId}:network-connector:*' + - Sid: Registry + Effect: Allow + Action: [dynamodb:GetItem, dynamodb:PutItem, dynamodb:UpdateItem, dynamodb:Scan] + Resource: !GetAtt TenantsTable.Arn + - Sid: SelfInvoke + Effect: Allow + Action: lambda:InvokeFunction + Resource: !Sub 'arn:aws:lambda:${AWS::Region}:${AWS::AccountId}:function:${ProjectName}-orchestrator' + + # ---------- Orchestrator Lambda ---------- + # Code comes from the bundled zip deploy.sh uploaded to S3 (boto3 + lambda-microvms model + # overlay + handler.py). Every env var is wired declaratively via GetAtt — no post-deploy CLI. + OrchestratorFn: + Type: AWS::Lambda::Function + Properties: + FunctionName: !Sub '${ProjectName}-orchestrator' + Runtime: python3.12 + Architectures: [arm64] + Handler: handler.handler + Role: !GetAtt OrchestratorRole.Arn + Timeout: 300 + MemorySize: 512 + Code: + S3Bucket: !Ref ArtifactBucketName + S3Key: !Ref OrchestratorCodeKey + Environment: + Variables: + TENANTS_TABLE: !Ref TenantsTable + EXEC_ROLE_ARN: !GetAtt MicroVMExecutionRole.Arn + INGRESS_CONNECTOR: !Sub 'arn:aws:lambda:${AWS::Region}:aws:network-connector:aws-network-connector:ALL_INGRESS' + EGRESS_CONNECTOR: !GetAtt EgressConnector.Arn + IMAGE_ARN: !GetAtt MicrovmImage.ImageArn + IMAGE_VERSION: !GetAtt MicrovmImage.LatestActiveImageVersion + GATEWAY_TOKEN: !Ref GatewayToken + BEDROCK_MODEL_ID: !Ref BedrockModelId + IDLE_REAP_SECONDS: !Ref IdleReapSeconds + + # ---------- API Gateway HTTP API → orchestrator (payload v2 proxy) ---------- + HttpApi: + Type: AWS::ApiGatewayV2::Api + Properties: + Name: !Sub '${ProjectName}-api' + ProtocolType: HTTP + Target: !GetAtt OrchestratorFn.Arn + + ApiInvokePermission: + Type: AWS::Lambda::Permission + Properties: + FunctionName: !Ref OrchestratorFn + Action: lambda:InvokeFunction + Principal: apigateway.amazonaws.com + SourceArn: !Sub 'arn:aws:execute-api:${AWS::Region}:${AWS::AccountId}:${HttpApi}/*/*' + + # ---------- EventBridge: sweeper every 10 min ---------- + SweeperRule: + Type: AWS::Events::Rule + Properties: + Name: !Sub '${ProjectName}-sweeper' + ScheduleExpression: rate(10 minutes) + State: ENABLED + Targets: + - Id: orchestrator + Arn: !GetAtt OrchestratorFn.Arn + Input: '{"_sweeper":true}' + + SweeperPermission: + Type: AWS::Lambda::Permission + Properties: + FunctionName: !Ref OrchestratorFn + Action: lambda:InvokeFunction + Principal: events.amazonaws.com + SourceArn: !GetAtt SweeperRule.Arn + +Outputs: + ApiEndpoint: + Description: Base URL; tenant webhooks are /tg/ + Value: !GetAtt HttpApi.ApiEndpoint + ArtifactBucket: + Value: !Ref ArtifactBucketName + MicrovmImageArn: + Value: !GetAtt MicrovmImage.ImageArn + MicrovmImageVersion: + Value: !GetAtt MicrovmImage.LatestActiveImageVersion + EgressConnectorArn: + Value: !GetAtt EgressConnector.Arn + EfsId: + Value: !Ref Efs + OrchestratorFunctionName: + Value: !Ref OrchestratorFn + TenantsTable: + Value: !Ref TenantsTable