Forem: szymon-szym

Run my own serverless service on a Raspberry Pi with Spin

szymon-szym — Tue, 14 Apr 2026 15:52:46 +0000

I want to launch my own serverless service running in my local network

Idea

As a huge fan of serverless computing, I thought it would be fun to run your own "serverless-like" environment locally and use it inside the home network.

A while ago, I started playing with the Spin framework. This is an awesome way to create applications in WASM and run them basically anywhere. Spin is an open-source, Cloud Native Foundation sandbox project.

WASM itself is an interesting technology, and it fits great into the serverless environment. The cool thing about it is that it runs in the sandbox and doesn't bring much execution overwhelm. There are a few languages that have some way to compile to WASM.

Part of the fun is that I am going to use my Raspberry Pi 3 as a server running Spin. I will "publish" my functions there. The expectation is that I will end up with my own energy-efficient serverless service running inside my home network.

Architecture

The code is available in the GitHub repository

Raspberry Pi is my server that runs Spin service and exposes the endpoint to call APIs. At this point, it is connected to my local network and can be reached only from it.

I develop an application on my local machine and deploy it by rsync over SSH.

The data for the app is persistent in the sqlite DB that runs on the Raspberry Pi

Goal

I create a book catalog that I can query for books I have. For this experiment, it will be a single endpoint that returns a book based on ID.

Application development

I start from installing Spin on my local machine
I install ready-to-use templates for Rust (there are also versions for Python, TypeScript, TinyGo) spin templates install --git https://github.com/spinframework/spin --update
The last thing would be to add WASM target to Rust toolchain: rustup target add wasm32-wasip1

Once it is done, let's create the app boilerplate. I run spin up and pick the http-rust template

Now I run spin build and spin up

All right! The app is running on the localhost and I can CURL it to see a dummy "Hello World"

Ok, let's write actual code. Spin abstracts away "server-ish" part, and working in it feels a bit like writing Lambda Functions. Components are defined in the spin.toml

#:schema https://schemas.spinframework.dev/spin/manifest-v2/latest.json

spin_manifest_version = 2

[application]
name = "books-spin"
version = "0.1.0"
authors = ["szymon"]
description = ""

[[trigger.http]]
route = "/book"
component = "books-spin"

[component.books-spin]
source = "target/wasm32-wasip1/release/books_spin.wasm"
allowed_outbound_hosts = []
[component.books-spin.build]
command = "cargo build --target wasm32-wasip1 --release"
watch = ["src/**/*.rs", "Cargo.toml"]

I have an HTTP trigger defined along with the path. There is also a build configuration for the component.

The handler code looks familiar. I start by defining the data model

// lib.rs
#[derive(Serialize, Debug)]
pub struct Book {
    pub title: String,
    pub author: String,
    pub year: u32,
}

Now the handler. Inside, I define a router and add handlers for specific methods.

// lib.rs
#[http_component]
fn handle_books_spin(req: Request) -> anyhow::Result<impl IntoResponse> {
    let mut router = spin_sdk::http::Router::new();

    router.add("/book", spin_sdk::http::Method::Get, api::get_book_by_id);

    println!("Handling request to {:?}", req.header("spin-full-url"));

    Ok(router.handle(req))
}

Finally, the function to get the given book. For now it returns some dummy data

// lib.rs

mod api {
    use spin_sdk::http::Params;

    use super::*;

    pub(crate) fn get_book_by_id(
        _req: Request,
        _params: Params,
    ) -> anyhow::Result<impl IntoResponse> {
        let dummy_book = Book {
            title: String::from("Dummy Book"),
            author: String::from("Dummy Author"),
            year: 2023,
        };

        Ok(Response::builder()
            .status(200)
            .header("content-type", "application/json")
            .body(serde_json::to_string(&dummy_book)?)
            .build())
    }
}

Let's test it with spin build && spin up

Add `SQLite`

One of the cool features of the Spin is that it wraps APIs for various features. One of them is an interface for the embedded SQLite Database.

I just need to add a list of all allowed SQLite databases to the component's definition.

#...

[component.books]
source = "books/target/wasm32-wasip1/release/books.wasm"
allowed_outbound_hosts = []
sqlite_databases = ["default"]

#...

I create a simple migration file

CREATE TABLE IF NOT EXISTS books (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    title TEXT NOT NULL,
    author TEXT NOT NULL,
    year INTEGER NOT NULL
);

INSERT INTO books (id, title, author, year) VALUES (1, 'Wieża Jaskółki', 'Andrzej Sapkowski', 2014)
ON CONFLICT(id) DO UPDATE SET title=excluded.title, author=excluded.author, year=excluded.year;

Now, it is enough to run Spin with the sqlite flag. The db file is created in the .spin directory

spin up --sqlite @migration.sql

For now, I keep connection inside a handler function. I am not sure if in the WASM environment, there is the idea of warm and cold execution. I need to check it, but for my small POC it doesn't really matter.

The updated handler function

// lib.rs
// ...
pub(crate) fn get_book_by_id(
        _req: Request,
        params: Params,
    ) -> anyhow::Result<impl IntoResponse> {
        println!("{:?}", &params);

        let id_param = params.get("id").ok_or(anyhow!("Missing id parameter"))?;

        let id = id_param.parse::<i64>()?;

        let query_params = [Value::Integer(id)];

        let connection = Connection::open_default()?;

        let rowset = connection.execute(
            "SELECT title, author, year FROM books WHERE id = ?",
            &query_params,
        )?;

        let books: Vec<Book> = rowset
            .rows()
            .map(|row| Book {
                title: row.get::<&str>("title").unwrap().to_string(),
                author: row.get::<&str>("author").unwrap().to_string(),
                year: row.get::<u32>("year").unwrap(),
            })
            .collect();

        let book = books.first().ok_or(anyhow!("Book not found"))?;

        Ok(Response::builder()
            .status(200)
            .header("content-type", "application/json")
            .body(serde_json::to_string(&book)?)
            .build())
    }
//...

Now I should be able to get a book by ID.

Nice.

Deploy to Raspberry Pi

I am using a Raspberry Pi 3 B+. After SSHing into it, I install the Spin. I don't need any tools for development, as the idea is to load compiled WASM files.

The flow I plan to use is the following: I create a systemd service to run my Spin server. Then I sync spin.toml, SQL migration, and built binaries using rsync. After each sync, I restart the service.

My service is defined in the spin-books.service

[Unit]
Description=Spin Books Service
After=network.target

[Service]
User=szymon
Group=szymon
WorkingDirectory=/home/szymon/projects/spin-books
ExecStart=/usr/local/bin/spin up --listen 0.0.0.0:3003 --sqlite @migration.sql
Restart=always

[Install]
WantedBy=multi-user.target

I sync this file to the /etc/systemd/system/ directory on the Raspberry Pi.

After running spin build on the local machine, I synced the config and created binaries:

rsync -avz --relative spin.toml migration.sql target/wasm32-wasip1/release/books_spin.wasm szymon@szymonpi3.local:~/projects/spin-books

Now I can run the service and make sure that my spin application is up and running on the Raspberry Pi

sudo systemctl daemon-reload
sudo systemctl start spin-books
sudo systemctl status spin-books

Looks good:

I can call the endpoint from my development machine:

Fantastic! I have my own WASM-based application on the Raspberry Pi running in my local network.

One more thing: I make sure it will start after rebooting by running sudo systemctl enable spin-books.service. Now every time I start the Raspberry Pi, my book service starts too.

Summary

When working on the blog post, I wanted to check how hard it would be to build a simple WASM-based backend application. It turned out that Spin makes it really easy. It can be easily run on the Raspberry Pi, which unlocks a fun experience of building your own serverless service.

The code is available in the GitHub repository

Safe AWS Lambda Durable Functions with XState

szymon-szym — Tue, 03 Mar 2026 18:57:28 +0000

Photo by Heather McKean on Unsplash

Define typesafe workflows running as AWS Lambda Durable Functions using XState

Code is available in this repository

Background

One of the most exciting announcements made during last re:Invent, at least for the serverless enthusiasts, was the introduction of Lambda Durable Functions.

At first glance, it appeared to me that Durable Functions might be a nice option for relatively simple workflows. If you need to define a complex workflow, you would probably lean towards AWS Step Functions, which allow creating full-blown state machines.

Problem

I am somehow biased when it comes to type safety in programming. I really appreciate the help from a compiler, and I feel much safer when the illegal state is unrepresentable.

That is one of the reasons I like to use AWS Step Functions. If the flow is created as a state machine, all possible transitions are strictly defined. Your business logic will never end up in a state that is not described in the state machine.

However, AWS Step Functions have their flaws. The main, from my perspective, is that state machine definition is created using Amazon States Language (ASL) in combination with JSON Path or JSONata. I can't expect any type safety there. In other words, if I need to implement any non-trivial change in the flow, I need to run it to see if it even works.

AWS Durable Functions introduced new possibilities. Now the flow can be defined using TypeScript, which should provide much more confidence compared to JSON-based ASL.

Goal

I would like to end up with a typed state machine definition that I could run on the AWS Durable Functions and benefit from native features, like observability and retries.

Experiment

When working on the side projects, I have been using the XState library. It allows defining state machines and using the actor model to strictly describe the business logic. Usually, XState is mentioned in the context of the FE, but the model is universal, so it can be used on the BE too.

If you want to learn more about the ideas behind XState, please check the official documentation

Implementation

Code is available in this repository

I create a simple flow that would perform two operations and return a message combined from their outputs.

As a starting point, I use a template of the Durable Function provided by SAM.

State Machine

XState is an amazing implementation of the state charts and the actor model. In my example, I barely scratch the surface of the available possibilities.

I define states and transitions. I also have actors to perform async operations. The idea is to wrap these operations in a durable context, so they can be checkpointed and retried by the Durable Functions runtime.

To achieve it, I am passing DurableContext object to the state machine's context so I can use it in my actors.

Actors

In the world of the states machines, actors are entities that perform some actions. I am aware that this might look a bit confusing if you haven't used XState before. But really, the idea is not that complicated.

I define functions to perform the actors' logic. In my case, they simply return the string. I also pass the DurableContext (from Durable Functions SDK) as an argument, so I can use it to create checkpoints:

const getSuccessMessage = async (
  durableContext: DurableContext,
): Promise<string> => {
  return durableContext.step("success_step", async () => {
    durableContext.logger.info("[machine] Executing success step");
    return Promise.resolve("success message from first step");
  });
};

For the step with retries, the function logic is the same:

const getMaybeMessage = async (
  durableContext: DurableContext,
): Promise<string> => {
  return durableContext.step(
    "maybeStep",
    async () => {
      durableContext.logger.info("[machine] Generating message");
      if (Math.random() < 0.4) {
        durableContext.logger.info("[machine] Generating success message");
        return Promise.resolve("success message from maybe step");
      } else {
        durableContext.logger.info("[machine] Generating failure message");
        return Promise.reject(new Error("Execution failed"));
      }
    },
    {
      retryStrategy: (error, attemptCount) => {
        if (attemptCount > 5) {
          return { shouldRetry: false };
        }
        const delay = Math.pow(2, attemptCount - 1);
        return { shouldRetry: true, delay: { seconds: delay } };
      },
    },
  );
};

The magic happens if the function returns Promise.reject. In that case, the execution is stopped, and it will be retired. This is exactly what we want.

To turn functions into actors, I use the fromPromise helper.

const getMaybeMessageActorLogic = fromPromise(
  async ({ input }: { input: { durableContext: DurableContext } }) =>
    getMaybeMessage(input.durableContext),
);

State Machine

Now let's define the flow. One of the most awesome features of the XState is that it goes well with TypeScript, allowing us to have strongly typed state machines. I use the .setup method to define the types expected in the machine

export const simpleMachine = (durableContext: DurableContext) => {
  return setup({
    actors: {
      getSuccessMessageActorLogic,
      getMaybeMessageActorLogic,
      startFromInitialStateLogic,
    },
    types: {
      context: {} as {
        durableContext: DurableContext;
        message: string;
      },
      output: {} as { message: string },
    },
  }).createMachine({
    // machine implementation
})

The setup looks inconspicuously but for me it is a game-changer. It provides strong types in the state machine definition.

The state machine definition looks exactly how you would expect:

export const simpleMachine = (durableContext: DurableContext) => {
  return setup({
    // setup definition
  }).createMachine({
    initial: "initial",
    context: {
      durableContext: durableContext,
      message: "",
    },
    output: ({ context }) => ({ message: context.message }),
    states: {
      initial: {
        invoke: {
          src: "startFromInitialStateLogic",
          input: ({ context }) => ({ durableContext: context.durableContext }),
          onDone: {
            target: "success_step",
          },
        },
      },
      success_step: {
        invoke: {
          src: "getSuccessMessageActorLogic",
          input: ({ context }) => ({ durableContext: context.durableContext }),
          onDone: {
            target: "maybe_step",
            actions: assign({
              message: ({ event, context }) => event.output,
            }),
          },
        },
      },
      maybe_step: {
        invoke: {
          src: "getMaybeMessageActorLogic",
          input: ({ context }) => ({ durableContext: context.durableContext }),
          onDone: {
            target: "end",
            actions: assign({
              message: ({ event, context }) =>
                event.output + " " + context.message,
            }),
          },
        },
      },
      end: {
        type: "final",
      },
    },
  });
};

In my simple case, the flow is linear and straightforward. I don't even use explicit events to move through states.

Type safety

As we have types defined in the setup stage, now, for example, we can't use the wrong actor's name.

And if we try, there is an error:

An actor's input and returned output are strongly typed. The same goes for the context, state machine output, and events, including their payload.

Visual representation

The thing I really like about AWS Step Functions is the fact that the state machine is presented in the console as a graph. The cool thing with the XState is that you can have the same experience.

You need to go to https://stately.ai/, which is the company behind XState, and open the visual editor. The experience is somewhat similar to the AWS Step Function. We can start creating workflows in the editor or paste the TypeScript code to have a flow rendered.

The visual editor is a proprietary tool and requires creating an account. For public projects, it is free.

Run on Durable Functions

In my case, I am using SAM, so I simply run sam build && sam deploy in the terminal.

Let's test the function in the console.

By checking the execution history, I can confirm that all steps were properly checkpointed. One of them failed and was retired. On the timeline, I can see that the second retry was performed according to the backoff strategy

The list of events

Summary

The result of my experiment is more than promising. XState provides a powerful framework for using state machines and the actor model. It helps define even very complex flows, with multiple nested machines acting as actors. It works smoothly with TypeScript, providing type safety for state machines.

AWS Durable Functions come with a handy SDK that makes it easy to use checkpoints and replies.

The described approach might be useful if you need a safe way to define complex workflows that are easy to test locally. I will definitely explore this direction further.

Streaming AI Agent response with AWS REST Gateway and Lambda Function

szymon-szym — Wed, 03 Dec 2025 13:20:19 +0000

Photo by Daniel Seßler on Unsplash

Recently, the number one use case for streaming responses is probably interactions with LLMs. It allows providing a better user experience in chat interfaces.

Now it is possible to stream a response from Lambda functions using the API REST Gateway. It is big news, as previously streaming response from AWS Lambda has been working in a limited way. It was possible only using the Function URL functionality, and it didn't fit into all use cases.

Goal

In this blog post, I will create a simple AWS Lambda that calls AWS Bedrock Model and streams the response back to the caller. I will use Rust to keep my Lambda small, fast and cheap.
The infrastructure is to be defined with AWS CDK.

Architecture

Lambda Function

I am using rig.rs as my AI framework. Let' start from defining main function

mod http_handler;
use http_handler::function_handler;
use lambda_runtime::{run, service_fn, tracing, Error};
use rig::client::completion::CompletionClientDyn;
use rig_bedrock::{client::Client, completion::AMAZON_NOVA_MICRO};

#[tokio::main]
async fn main() -> Result<(), Error> {
    tracing::init_default_subscriber();

    let aws_config = aws_config::from_env().region("us-east-1").load().await;

    let aws_client = aws_sdk_bedrockruntime::Client::new(&aws_config);

    let agent = Client::from(aws_client)
        .agent(AMAZON_NOVA_MICRO)
        .preamble("be concise")
        .temperature(0.5)
        .build();

    run(service_fn(|ev| function_handler(agent.clone(), ev))).await
}

Now the handler. It starts almost identically as a regular one. The main difference is that Response and Body types are imported from lambda_runtime::streaming module.


#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct PromptRequest {
    prompt: String,
}

pub(crate) async fn function_handler(
    agent: Agent<CompletionModelHandle<'static>>,
    event: LambdaEvent<ApiGatewayProxyRequest>,
) -> Result<Response<Body>, Error> {
    let (mut tx, rx) = channel();

    println!("prompt: {:?}", &event.payload);

    let body = event
        .payload
        .body
        .ok_or(Box::new(Error::from("failed to read body")))
        .expect("failed to read body");

    let prompt_request: PromptRequest =
        serde_json::from_str(&body).map_err(|_| Error::from("failed to parse body"))?;

    let mut bedrock_stream = agent.stream_prompt(prompt_request.prompt).await;
// ...

The crucial thing is that we are creating a transmitter tx and a receiver rx for handling stream operations.

In the rig library, agent.stream_prompt returns a stream of MultiTurnStreamItem. This is the enum:

// rig crate
pub enum MultiTurnStreamItem<R> {
    /// A streamed assistant content item.
    StreamAssistantItem(StreamedAssistantContent<R>),
    /// A streamed user content item (mostly for tool results).
    StreamUserItem(StreamedUserContent),
    /// The final result from the stream.
    FinalResponse(FinalResponse),
}

The idea is that having strongly typed streamed items, I can easily pattern match when receiving "something" from the stream, and fully control the logic. In my case, I will stream text from AssistantItem.

The tricky part is how to "translate" the streamed response from rig to lambda format. To make it possible, we need to spawn a new tokio thread, so the execution of the function won't finish before receiving the response from the Bedrock model.
Then I will read from bedrock_stream using the next() method from the futures crate.

tokio::spawn(async move {
        while let Some(chunk_result) = bedrock_stream.next().await {
            match chunk_result {
                Ok(chunk) => match chunk {
                    MultiTurnStreamItem::StreamAssistantItem(assistant_item) => {
                        match assistant_item {
                            StreamedAssistantContent::Text(text) => {
                                if tx
                                    .send_data(Bytes::from(text.text().as_bytes().to_vec()))
                                    .await
                                    .is_err()
                                {
                                    error!("error client disconnected");
                                    break;
                                }
                            }
                            StreamedAssistantContent::Reasoning(_reasoning) => {
                                if tx
                                    .send_data(Bytes::from("--- thinking ---".as_bytes()))
                                    .await
                                    .is_err()
                                {
                                    error!("error client disconnected");
                                    break;
                                }
                            }
                            _ => {
                                info!("processing tools")
                            }
                        }
                    }
                    MultiTurnStreamItem::FinalResponse(_agent_response) => {
                        info!("final response");
                    }
                    _ => {
                        info!("-- processing")
                    }
                },
                Err(_) => {
                    error!("stream error");
                    break;
                }
            }
        }
    });

At the end of the execution, I return rx

Ok(Response::from(rx))

Infrastructure

Setup is basically identical to the buffered version of lambda integration for the REST API Gateway. The only thing is a new responseTransferMode property set as STREAM. Make sure that you are using the newest version of the cdk-lib

import * as cdk from "aws-cdk-lib";
import * as apigw from "aws-cdk-lib/aws-apigateway";
import * as iam from "aws-cdk-lib/aws-iam";
import { Construct } from "constructs";
import { RustFunction } from "cargo-lambda-cdk";

export class StreamingInfraStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props);

    const handler = new RustFunction(this, "Streaming Function", {
      manifestPath: __dirname + "/../../streaming_function/Cargo.toml",
    });

    handler.addToRolePolicy(
      new iam.PolicyStatement({
        effect: iam.Effect.ALLOW,
        actions: [
          "bedrock:InvokeModel",
          "bedrock:InvokeModelWithResponseStream",
        ],
        resources: [`arn:aws:bedrock:${this.region}::foundation-model/*`],
      }),
    );

    const api = new apigw.RestApi(this, "Streaming API", {
      restApiName: "Streaming API",
      description: "API for streaming data",
      deployOptions: {
        stageName: "prod",
      },
    });

    const promptResource = api.root.addResource("prompt");

    const promptIntegration = new apigw.LambdaIntegration(handler, {
      responseTransferMode: apigw.ResponseTransferMode.STREAM,
    });

    const promptResourceMethod = promptResource.addMethod(
      "POST",
      promptIntegration,
    );

    new cdk.CfnOutput(this, "ApiEndpointUrl", {
      value: api.url,
      description: "Base URL of the REST API Gateway",
      exportName: "MyRestApiEndpoint",
    });
  }
}

Testing

To test the solution, I can use curl and see the response streamed back

The response is almost instant, but we can still see that the text is rendered gradually.

Summary

Today, the API REST Gateway can stream responses from Lambda functions. It makes it much easier to utilize serverless infrastructure to provide users with output from LLMs.

Using Rust helps keep the latency at a low level, with the init duration around 100ms for a cold start. As a strongly typed language, it also makes it easier to control the flow of the streaming.

Deploy Rust Agent to AWS AgentCore Runtime with GitHub actions

szymon-szym — Sun, 23 Nov 2025 14:44:08 +0000

Photo by Brian Cockley on Unsplash

AgentCore Runtime works out of the box with Python frameworks. It also allows for deploying agents created with other languages. Using Rust to build an agent sounds like a nice excuse to explore this territory

Goal

For this blog post, I would like to:

deploy containerised Rust application to AgentCore Runtime
define all infrastructure as code (AWS CDK in my case)
test agent locally
build and deploy in CI/CD pipeline (GitHub Actions)

The agent itself will be simple, as I want to focus on the deployment process.

Architecture

The code is available IN THE REPOSITORY (on the 01-initial branch)

Agent

The agent in the AgentCore Runtime is nothing but a web application that exposes two endpoints: /ping and /invocations

For my project, I use the axum framework. Let's define request and response types:

// main.rs
//...
#[derive(Serialize)]
pub struct StatusResponse {
    #[serde(rename = "status")]
    status_msg: String,
}

#[derive(Debug, Serialize, Deserialize)]
pub struct InvocationsRequest {
    prompt: String,
}

#[derive(Serialize)]
pub struct InvocationsResponse {
    message: String,
}

//...

/ping endpoint is expected to return {"status": "healthy"}
/invocations will accept an object with prompt and return message.

/Invocations

The handler looks like this:

#[debug_handler]
async fn invocations(
    State(state): State<AppState>,
    Json(payload): Json<InvocationsRequest>,
) -> Json<InvocationsResponse> {
    let prompt = payload.prompt;

    let response = state.agent.prompt(prompt).await.unwrap();

    let response = InvocationsResponse { message: response };
    Json(response)
}

I use axum extractors for getting a state with a prepared rig agent, and for working with json

Main function

I initialise the AWS SDK client to use it with rig, and register handlers

// main.rs
// ...
#[tokio::main]
async fn main() {
    tracing_subscriber::fmt().json().init();

    let aws_config = aws_config::from_env().region("us-east-1").load().await;

    let bedrock_runtime_client = aws_sdk_bedrockruntime::Client::new(&aws_config);

    let client = Client::from(bedrock_runtime_client);

    let agent = client
        .agent(AMAZON_NOVA_PRO)
        .preamble("You are helpful assistant. Respond only with the answer. Be concise.")
        .build();

    let state = AppState { agent: agent };

    let app = Router::new()
        .route("/ping", get(ping))
        .route("/invocations", post(invocations))
        .with_state(state);

    let listener = TcpListener::bind("0.0.0.0:8080").await.unwrap();
    axum::serve(listener, app).await.unwrap();
}
// ...

Now I can test my agent locally.
I can run cargo run and CURL the prompt:

Deployment

With the agent working locally is now a time to start working on the deployment process.

From here on, I iterated a lot to figure out how to make it work. To keep things clear, I won't describe all my struggles in chronological order. Some things were obvious, while others turned out to be rather tricky.

Dockerfile

AgentCore Runtime supports only ARM images. I will build the artefact using GitHub Actions ARM runner, so I don't need any emulation.

FROM rust:1.91 AS builder

RUN rustup target add aarch64-unknown-linux-gnu

WORKDIR /app

COPY . .

RUN cargo build --release --target aarch64-unknown-linux-gnu

FROM ubuntu:24.04 AS runtime

RUN apt-get update && apt-get install -y --no-install-recommends ca-certificates libssl3 && update-ca-certificates && rm -rf /var/lib/apt/lists/*

WORKDIR /app

COPY --from=builder /app/target/aarch64-unknown-linux-gnu/release/agent .

ENV AWS_REGION=us-east-1 \
    AWS_DEFAULT_REGION=us-east-1 \
    RUNNING_ENV=agentcore

RUN useradd -m appuser
USER appuser

CMD ["./agent"]

For my application to work, I need some system dependencies. That's why I am installing libssl3. Initially, I tried to use a smaller image for the runner, but I ended up with a default ubuntu to have a new version of GLIBC available. For now, I finished with a ~40MB image in the ECR, which is significantly smaller than the default Python-based one anyway.

Infratructure

The tricky part of preparing the infrastructure for AgentCore Runtime is that specific elements need to be ready in a specific order. What I mean is that I am not able to host the agent until the actual image is uploaded to ECR. In other words, initial deployment needs to be thought through.

There are a few ways to tackle this problem. You might, for example, use custom resources in CloudFormation. I decided to keep things simple and split the infrastructure into two stacks. Once the ECR repository is created, I build a Docker image and then deploy AgentcoreRuntime.

The hardest part is to have the first deployment right. Later on, it is much easier, as most of the components are already deployed.

Deployment pipeline

Let's go through the GitHub Actions workflow step by step

Set up AWS credentials and install CDK

# .github/workflows/deploy.yaml
name: Deploy Agent

on:
  push:
    branches: ["main"]

env:
  AWS_REGION: ${{ vars.AWS_REGION }}
  AWS_ROLE: ${{ secrets.AWS_ROLE }}

permissions:
  id-token: write
  contents: read

jobs:
  deploy-agent:
    runs-on: ubuntu-24.04-arm

    defaults:
      run:
        working-directory: agent

    steps:
      - name: Checkout
        uses: actions/checkout@v4

      - name: Configure AWS credentials
        uses: aws-actions/configure-aws-credentials@v4
        with:
          role-to-assume: ${{ env.AWS_ROLE }}
          aws-region: ${{ env.AWS_REGION }}

      - name: Install CDK
        run: |
          npm install -g aws-cdk
# ...

I have a role created in IAM to grant GitHub access to the AWS. The process of creating it is described here: LINK
I store the role ARN in the GitHub secrets.

Deploy ECR repository

The repository definition:

import * as cdk from "aws-cdk-lib";
import { Construct } from "constructs";
import * as ecr from "aws-cdk-lib/aws-ecr";

export class AcInfraStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props);

    const acRepository = new ecr.Repository(this, "RustAgentRepository", {
      repositoryName: "agentcore-rust-agent-repo",
    });

    new cdk.CfnOutput(this, "ECRRepositoryURI", {
      value: acRepository.registryUri,
    });

    new cdk.CfnOutput(this, "ECRRepositoryName", {
      value: acRepository.repositoryName,
    });
  }
}

I could create exports for repository name and URI, but I am not a huge fan of exports, so I just return them as outputs and extract in the GitHub action.

# ...
- name: Deploy ECR
        working-directory: aws/ac-infra
        run: |
          npm i
          cdk deploy -O cdk-ecr-outputs.json --require-approval never

      - name: Parse ECR CDK outputs
        id: cdk_ecr_outputs
        working-directory: aws/ac-infra
        run: |
          REPO_NAME=$(jq -r '.RustAgentAgentcoreECRStack.ECRRepositoryName' cdk-ecr-outputs.json)
          REPO_URI=$(jq -r '.RustAgentAgentcoreECRStack.ECRRepositoryURI' cdk-ecr-outputs.json)

          echo "repo_name=$REPO_NAME" >> $GITHUB_OUTPUT
          echo "repo_uri=$REPO_URI" >> $GITHUB_OUTPUT

Build the image

Once the Dockerfile is prepared, there is not much magic in the build step. The best part is that there are ARM runners available on GitHub for free for public repositories. Pretty nice!

I use a dynamic repository name and URI stored after the previous steps

# ...
- name: Docker login to ECR
        run: |
          aws ecr get-login-password --region $AWS_REGION \
            | docker login \
              --username AWS \
              --password-stdin ${{ steps.cdk_ecr_outputs.outputs.repo_uri }}

      - name: Build ARM64 Docker image
        run: |
          docker build \
            -t ${{ steps.cdk_ecr_outputs.outputs.repo_name }}:latest \
            .

      - name: Tag image
        run: |
          docker tag ${{ steps.cdk_ecr_outputs.outputs.repo_name }}:latest \
            ${{ steps.cdk_ecr_outputs.outputs.repo_uri }}/${{ steps.cdk_ecr_outputs.outputs.repo_name }}:latest

      - name: Push to ECR
        run: |
          docker push \
          ${{ steps.cdk_ecr_outputs.outputs.repo_uri }}/${{ steps.cdk_ecr_outputs.outputs.repo_name }}:latest

Deploy AgentCore Runtime

With the ECR repository in place and the image uploaded, I can now deploy AgentCore Runtime. For the initial version, I stick to the simple public endpoint with IAM authorisation.

// ...
const AGENT_NAME = "rust_agent";

    const repositoryName = this.node.tryGetContext("REPO_NAME");
    const repositoryURI = this.node.tryGetContext("REPO_URI");

    const runtimeRole = new iam.Role(this, "AgentCoreRustAgent", {
      assumedBy: new iam.ServicePrincipal("bedrock-agentcore.amazonaws.com", {
        conditions: {
          StringEquals: {
            "aws:SourceAccount": this.account,
          },
          ArnLike: {
            "aws:SourceArn": `arn:aws:bedrock-agentcore:${this.region}:${this.account}:*`,
          },
        },
      }),
    });

// -- a lot of "runtimeRole.addToPolicy" to fulfil requirements for execution role --

    const agentRuntime = new agentcore.CfnRuntime(this, "RustAgent", {
      agentRuntimeArtifact: {
        containerConfiguration: {
          containerUri: `${repositoryURI}/${repositoryName}:latest`,
        },
      },
      agentRuntimeName: AGENT_NAME,
      networkConfiguration: {
        networkMode: "PUBLIC",
      },
      roleArn: runtimeRole.roleArn,
    });

    agentRuntime.node.addDependency(runtimeRole);

    new cdk.CfnOutput(this, "RustAgentId", {
      value: agentRuntime.attrAgentRuntimeId,
    });

    new cdk.CfnOutput(this, "AgentRuntimeRoleArn", {
      value: runtimeRole.roleArn,
    });
  }

Again, I use the simple output to pass variables down the GitHub pipeline.

# ...
- name: Deploy AgentCore Runtime
        working-directory: aws/ac-runtime
        run: |
          npm i
          cdk deploy -O cdk-runtime-outputs.json --require-approval never \
          --context REPO_URI=${{ steps.cdk_ecr_outputs.outputs.repo_uri }} \
          --context REPO_NAME=${{ steps.cdk_ecr_outputs.outputs.repo_name }}
# ...

Bump agent version

AgentCore Runtime will pull the current image from the repository from time to time. I am not sure how often, though. Probably the safest way is to bump the Agent version to make sure that the current image is in use, and to track changes across versions.

I run AWS CLI to update the Runtime using variables returned from the deployment step

# ...
- name: Parse Runtime CDK outputs
        id: cdk_runtime_outputs
        working-directory: aws/ac-runtime
        run: |
          AGENT_RUNTIME_ROLE=$(jq -r '.AcRuntimeStack.AgentRuntimeRoleArn' cdk-runtime-outputs.json)
          AGENT_RUNTIME_ID=$(jq -r '.AcRuntimeStack.RustAgentId' cdk-runtime-outputs.json)

          echo "agent_runtime_role=$AGENT_RUNTIME_ROLE" >> $GITHUB_OUTPUT
          echo "agent_runtime_id=$AGENT_RUNTIME_ID" >> $GITHUB_OUTPUT

      - name: Update Agent version
        run: |
          aws bedrock-agentcore-control update-agent-runtime \
          --agent-runtime-id ${{ steps.cdk_runtime_outputs.outputs.agent_runtime_id }} \
          --agent-runtime-artifact "{
                  \"containerConfiguration\": {
                    \"containerUri\": \"${{ steps.cdk_ecr_outputs.outputs.repo_uri }}/${{ steps.cdk_ecr_outputs.outputs.repo_name }}:latest\"
                  }
                }" \
          --role-arn ${{ steps.cdk_runtime_outputs.outputs.agent_runtime_role }} \
          --network-configuration '{ "networkMode": "PUBLIC" }' \
          --region ${{ env.AWS_REGION }}

Finding out

When I merge this code into the main, it will deploy the whole environment and bump the agent version. This is pretty cool. However, the agent itself will... fail in the runtime due to missing permissions.

MicroVM Metadata Service (MMDS)

AgentCore Runtime uses MMDS to get credentials for the given actor. It is similar to the Instance Metadata Service on EC2. The bottom line is that developers don't need to worry about how it works under the hood, as a AWS SDK covers the authentication part.
That's why it works in Python out of the box.

However, the Rust AWS SDK fails to authenticate on the AgentCore Runtime, which is a bummer.

I didn't really find the answer in the documentation, but this is what I understood. Rust SDK is new and it supports only IMDS v2 authorising schema. It is understandable, as on EC2, we shouldn't use v1 anymore. Other libraries, like botocore, still have a fallback to IMDS v1 if v2 is not available.

From what I saw, the MicroVM Metadata Service uses the same schema as IMDS v1. That's why it fails on Rust, but works on Python.

To fix this issue, I just needed to create my own CredentialsProvider that would follow the schema of IMDS v2. The code is in the repo, but it took me a while to figure this out.

Resources provisioning

I struggled a bit with creating the AgentCore Runtime for the first time using CDK. CloudFormation under the hood runs validation when creating a new runtime. It looks like we need to explicitly say that CloudFormation should wait for the IAM role to be created before triggering provisioning new Runtime. Otherwise, creation will fail. In CDK, this is done by adding agentRuntime.node.addDependency(runtimeRole);

Testing

After many tries, I finally got it working for the fresh deployment

Agent is deployed fully from the CI/CD pipeline.

Let's test it:

Concise and without feelings. Just how I like it

Now let's update the system prompt and push the changes:

Runtime is updated:

Next steps

In the near future I want to check performance and memory/CPU usage. I run some load tests to see how AgentCore scales.

It will be interesting to utilise other AgentCore services like Memory or Identity for the Rust project.

Summary

AS a result of this blog post, I ended up with

the agent written in Rust, containerised and deployed to AgentCore runtime
IaC definition for AWS resources
automated CI/CD pipeline working in GitHub Actions

Agent conversation context with AWS Bedrock AgentCore Memory

szymon-szym — Mon, 03 Nov 2025 11:00:34 +0000

Photo by Photoholgic on Unsplash

You can seamlessly hook up the conversation context in the stateless agentic environment with AWS Bedrock AgentCore Memory

Goal

I want to create a customer support agent (how original), that is capable of holding the conversation context, and can recall main facts about the user in for example few days later.

# What I am building

For testing purposes, I don't need to build the UI, but the idea is that the user can interact with the agent by sending an HTTP request. Authentication is done using the OAuth flow.
The agent understands the context of the request and reaches out to the AgentCore Memory service. It uses the fetched information to enrich the context of the conversation.

The whole code is available IN THIS REPOSITORY

Implementation

I am using AWS AgentCore Runtime to host the agent, and AWS AgentCore Memory to store facts about users. For the user authentication, I have a Cognito user pool with a configured app client.

When building this POC, I reached out for agentcore starter toolkit, which is a perfect tool for the job.

I use Strands as my agentic framework, mostly because it simplifies integration with the AgentCore environment, e.g., managing memory hooks automatically.

Cognito

I create a Cognito user pool for managing user interactions with the agent. Any provider might be used for this, as we will need only the discovery URI and client ID. I define infrastructure with AWS CDK

import * as cdk from "aws-cdk-lib";
import { Construct } from "constructs";
import * as cognito from "aws-cdk-lib/aws-cognito";

export class AwsStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props);

    const userPool = new cognito.UserPool(this, "AgentCoreRuntimeUserPool", {
      selfSignUpEnabled: true,
      autoVerify: { email: true },
      signInAliases: { email: true },
      standardAttributes: {
        email: { required: true, mutable: true },
      },
    });

    const domainPrefix = "test-ac-pool";

    const domain = new cognito.UserPoolDomain(
      this,
      "AgentCoreRuntimeUserPoolDomain",
      {
        userPool,
        cognitoDomain: {
          domainPrefix,
        },
      },
    );

    const userPoolClient = new cognito.UserPoolClient(
      this,
      "AgentCoreRuntimeUserPoolClient",
      {
        userPool,
        generateSecret: true,
        authFlows: {
          userPassword: true,
          adminUserPassword: true,
          userSrp: true,
        },
        oAuth: {
          flows: {
            authorizationCodeGrant: true,
            implicitCodeGrant: true,
          },
          scopes: [
            cognito.OAuthScope.EMAIL,
            cognito.OAuthScope.OPENID,
            cognito.OAuthScope.PROFILE,
          ],
          callbackUrls: ["https://example.com/callback"],
          logoutUrls: ["https://example.com/logout"],
        },
      },
    );

    const discovery = ` https://cognito-idp.us-east-1.amazonaws.com/${userPool.userPoolId}/.well-known/openid-configuration`;

    new cdk.CfnOutput(this, "Discovery URI", {
      value: discovery,
      description: "The discovery URI for the user pool",
    });

    new cdk.CfnOutput(this, "Client ID", {
      value: userPoolClient.userPoolClientId,
      description: "The ID of the user pool client",
    });
  }
}

For now, callback URLs are dummy, as I won't connect the UI yet.

After running cdk deploy, I get the discovery URL and client ID

Agent

I create new directory and run uv init. AgentCore toolkit works with uv out of the box.

I add dependencies needed using uv add <package>
My pyproject.toml looks like this:

[project]
name = "bank-support-agent"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.11"
dependencies = [
    "bedrock-agentcore>=1.0.5",
    "bedrock-agentcore-starter-toolkit>=0.1.28",
    "boto3>=1.40.64",
    "python-dotenv>=1.2.1",
    "strands-agents>=1.14.0",
    "strands-agents-tools>=0.2.13",
]

I start by defining some constants and initializing BedrockAgentCoreApp

# main.py
# -- imports --
REGION = "us-east-1"
MODEL_ID = "amazon.nova-pro-v1:0"
MEMORY_ID = os.getenv("BEDROCK_AGENTCORE_MEMORY_ID")

app = BedrockAgentCoreApp()

session = Session(region_name=REGION)

model = BedrockModel(model_id=MODEL_ID, boto_session=session)

# ...

For simplicity's sake, I will keep everything in the one function. Moving forward, I would extract at least initialization part. AgentCore comes with a handy decorator to define the entry point

# ...

@app.entrypoint
def invoke(payload, context):

# ...

I extract actor_id session_id and prompt from the request payload and context.

# ...
    actor_id = headers.get(
        "x-amzn-bedrock-agentcore-runtime-custom-actorid", "default_actor"
    )

    session_id = getattr(context, "session_id", None)

    prompt = payload.get("prompt", "")
# ...

Now I create the Memory config. I opted for a default setup, but AgentCore Memory is highly configurable. We might extend built-in strategies with overrides or use our custom strategy. This is useful, e.g., if we need to collect only data about a specific business topic.

The most important properties for managing memories are session_id and actor_id. Session ID is responsible for the short-term memory, for understanding the current conversation context. The AgentCore Runtime is stateless, so it won't be possible without some storage mechanism. Actor ID is responsible for identifying the caller - user, or other agent, and for getting long-term memories related to the given persona. In our case, we will use built-in strategies for storing facts and preferences.

As I mentioned before, the Strands framework comes with ready-to-use integrations:

# ...

session_manager = None
    if MEMORY_ID:
        session_manager = AgentCoreMemoryConfig(
            memory_id=MEMORY_ID,
            session_id=session_id or "default",
            actor_id=actor_id,
            retrieval_config={
                f"/users/{actor_id}/facts": RetrievalConfig(
                    top_k=3, relevance_score=0.5
                ),
                f"/users/{actor_id}/preferences": RetrievalConfig(
                    top_k=3, relevance_score=0.5
                ),
            },
        )
        session_manager = AgentCoreMemorySessionManager(session_manager, REGION)

# ...

The next step would be to initiate the agent and to respond to the prompt:

 # ...
agent = Agent(
        model=model,
        session_manager=session_manager,
        system_prompt="""
        You are a helpful assistant. Your task is to support bank clients with their queries and concerns.
        Stay cohesive and return short answers.
    """,
        tools=[],
    )

    result = agent(prompt)

    return {"response": result.message.get("content", [{}])[0].get("text", str(result))}

As you see, it's a pretty basic agent definition.

And at the end, let's trigger the app

# ...

if __name__ == "__main__":
    app.run()

Prepare AgentCore config

Now it is time for the agentcore toolkit magic.

In the agent directory, I run uv run agentcore configure -e main.py -r us-east-1

Tool properly detects pyproject.toml. I let it create execturion role and ECR repository.

When it comes to Auth configuration, I enter my discovery URL ...

.. and client ID. Leaving audience empty

I allow X-Amzn-Bedrock-AgentCore-Runtime-Custom-ActorId and Authorization headers to be passed with the request. Finally I choose "yes" for creating memory.

Now the .bedrock_agentcore.yaml is create in root agent directory, and .bedrock_agentcore/bank_customer_service/Dockerfile is ready to be deployed.

Provide cloud resources

I run

uv run agentcore launch

The tool takes care of creating all necessare resources. It also triggers CodeBuild job to build the agent and push the image to ECR. Once it is built, it will create agent in the runtime, or will update the existing one with the new version

Once deployment is ready, I can see, that .bedrock_agentcore.yaml is updated with information about created cloud resources

Testing

There is a convenient method to test agents, using invoke command, e.g.

agentcore invoke '{"prompt": "Hello"}'

However, it won't work for us:

Getting Cognito token

This is not the most exciting part, but until there is no UI to log in, it might be problematic. To be able to get a token, I perform a few actions.

I need a secret hash to be used for creating, confirming, and getting a token using aws cli

echo -n "test_user@example.com<client_id>" | openssl dgst -sha256 -hmac <client_secret> -binary | openssl enc -base64

Now I can create a new user and confirm creation

aws cognito-idp sign-up --client-id <client_id> --username test_user@example.com --password Test1234! --region us-east-1 --secret-hash <hash>
aws cognito-idp admin-confirm-sign-up --user-pool-id <pool_id> --username test_user@example.com --region us-east-1

Finally, I can get a token with the newly created user:

aws cognito-idp admin-initiate-auth --user-pool-id <pool_id> --client-id <client_id> --auth-flow ADMIN_USER_PASSWORD_AUTH --auth-parameters USERNAME=test_user@example.com,PASSWORD='Test1234!',SECRET_HASH=<hash> --region us-east-1

Local test with Bruno

For development, it is most convenient to test the agent locally, without deploying every change to the cloud.

In the code, I rely on MEMORY_ID passed as an environment variable, so for local testing I create a .env file and I read it using dotenv. The Memory ID can be found in the .bedrock_agentcore.yaml

Since I am eventually going to use my agent from UI, I want to test it with HTTP request, instead of using agentcore invoke command.

I set headers x-amzn-bedrock-agentcore-runtime-custom-actorid for actor ID, and x-amzn-bedrock-agentcore-runtime-session-id for session ID.

To start local server: uv run main.py

And I can send a request. Looks good:

I can see in the logs that memory is loaded. I can test it locally, but I want to hit the live endpoint.

Testing live endpoint

To call our agent, we need to send a POST request to the endpoint:
https://bedrock-agentcore.<REGION>.amazonaws.com/runtimes/<URL_ENCODED_AGENT_ARN>/invocations?qualifier=DEFAULT

Agent ARN can be found in the .bedrock_agentcore.yaml file, or in the console.

I set the ActorId header as user_tech and I shared some insights with the agent:

Ok, it works fine as part of the same session_id. Maybe it doesn't look amazing, but consider that the AgentCore runtime is stateless.

Now let's try changing session_id

and ask the same question:

Great!

Let's now update session_id and actor_id to simulate a different user:

Nice :)

Summary

Using the AgentCore Memory service solves the context issues for stateless agent flows. With the session ID and actor ID, we are able to help the agent get relevant information and make the conversation more natural and less frustrating for the user.

Built-in strategies for storing memories are powerful. Strategies are also highly customizable. We can create our own logic for storing information.

AgentCore Memory is not a replacement for RAG. RAG is still relevant for getting domain-specific insights and general data. AgentCore Memory helps with keeping the context of the conversation related to the specific actor.

I didn't try other frameworks, but when using Strands, AgentCore Memory integration is seamless. I didn't need to set up any hooks and modify the code. It was enough to create a MemoryManager config and pass it to the agent.

Serverless Semantic Search: A Hands-On Guide to AWS S3 Vector Buckets with Rust

szymon-szym — Thu, 28 Aug 2025 11:47:33 +0000

Photo by Jeff Kingma on Unsplash

Introduction

In the AI era, a semantic search is an important tool to have under the belt. There are many options for storing embeddings, and there is a place for even more on the market. The recent release of the S3 Vector Bucket looks interesting as a fully serverless option for storing and querying vectors.

Goal

In this blog post, I will build an HTTP endpoint to query books by their descriptions. The request will also include an optional category field, so callers can filter only results from the given genre. I would like to check if the S3 Vector can offer an acceptable latency to be used in a user-facing service.

Architecture

Creating a bucket

CODE IS AVAILABLE IN THIS REPO

As S3 Vector storage is in preview, there is no CloudFormation support yet. A bucket and index can be created using the console. I followed the user guide and created a new bucket. Then I added the books index inside.

When creating an index, we need to decide upfront which type of distance metric we are going to use - Cosine or Euclidean. They utilize a different logic to look for the nearest neighbors. It is important, as we can't change these settings later on. In real life, we would probably create two buckets and test search on the data sample to find out which approach is better for our case. For this blog, I picked cosine.

As per vectors' dimensions, I stuck to 1024, as I planned to use Titan Text V2 for generating embeddings.

Populate data

As a datasource, I took books dataset on Kaggle. I used the first 1k rows.

AWS provides a handy CLI tool to populate vectors; however, I needed to build integrations for my lambda handler anyway, so I created the uploader script in Rust

When prototyping with Rust, I like to use unwrap everywhere and implement proper error handling later on. You were warned.

Let's start by defining domain types:

// rust_app/common/src/models.rs

use serde::{Deserialize, Serialize};

#[derive(Deserialize, Serialize, Debug, Clone)]
    pub struct Book {
        pub isbn13: u128,
        pub title: String,
        pub authors: String,
        pub categories: String,
        pub description: String,
    }

#[derive(Debug, Serialize)]
pub struct BookWithEmbeddings {
    pub item: Book,
    pub embeddings: Vec<f32>
}

In the uploader script, I read the csv file stored locally, and I get embeddings from Titan

// rust_app/vector_updater/src/main.rs

//...

let mut reader = csv::Reader::from_path("books2.csv").unwrap();

    let books: Vec<Book> = reader.deserialize()
        .map(|res| res.unwrap())
        .collect();

    println!("extracted {} books", books.len());

    // Initialize AWS config

    let config = aws_config::from_env()
        .region("us-east-1")
        .load()
        .await;

    let embeddings_svc = EmbeddingsService::new(&config).await;

    let books_for_upload = books.iter()
        .filter(|b| !b.description.is_empty())
        .cloned()
        .collect::<Vec<_>>();

    println!("getting embeddings for {} books", books_for_upload.len());


    let embeddings = books_for_upload
        .iter()
        .map(|book| get_book_with_embeddings(&embeddings_svc, book))
        .collect::<Vec<_>>();

    let books_with_embeddings = stream::iter(embeddings)
        .buffer_unordered(MAX_CONCURENCY)
        .collect::<Vec<_>>()
        .await;

    println!("{:?}", books_with_embeddings.len());

Embeddings are taken from Bedrock APIs using the following method

// rust_app/common/src/embeddings_svc.rs

pub async fn get_embeddings(&self, query: String) -> TitanResponse {

        let body = json!({
            "inputText": query
        });

        let body_blob = Blob::new(serde_json::to_vec(&body).unwrap());

        let response = self.client
            .invoke_model()
            .model_id("amazon.titan-embed-text-v2:0")
            .body(body_blob)
            .send()
            .await
            .unwrap();

        let bytes = response.body().clone().into_inner();

        let result: TitanResponse = serde_json::from_slice(&bytes).unwrap();

        result

    }

And now, once I have the books with generated embeddings, I upload data to S3 Vector Bucket

// rust_app/vector_updater/src/main.rs
let s3_vectors_client = aws_sdk_s3vectors::Client::new(&config);

    let put_vectors_calls = books_with_embeddings
        .iter()
        .map(|be| upload_vectors(&s3_vectors_client, be))
        .collect::<Vec<_>>();

    println!("Uploading {} records to S3", put_vectors_calls.len());

    let _ = stream::iter(put_vectors_calls)
        .buffer_unordered(MAX_CONCURENCY)
        .collect::<Vec<_>>()
        .await;

The upload_vectors method

// rust_app/vector_updater/src/main.rs

pub async fn upload_vectors(s3vectors_client: &aws_sdk_s3vectors::Client, book_with_embeddings: &BookWithEmbeddings) -> () {

    let mut metadata = HashMap::new();

    metadata.insert("source_text".to_string(), Document::String(book_with_embeddings.item.description.clone()));
    metadata.insert("categories".to_string(), Document::String(book_with_embeddings.item.categories.clone()));
    metadata.insert("title".to_string(), Document::String(book_with_embeddings.item.title.clone()));
    metadata.insert("authors".to_string(), Document::String(book_with_embeddings.item.authors.clone()));

    let vector_input =  
            PutInputVector::builder()
                .key(book_with_embeddings.item.isbn13.to_string())
                .data(VectorData::Float32(book_with_embeddings.embeddings.clone()))
                .metadata(Document::Object(metadata))
                .build()
                .unwrap();

    let _result = s3vectors_client
        .put_vectors()
        .vector_bucket_name(S3_NAME)
        .index_name(INDEX_NAME)
        .vectors(vector_input)
        .send()
        .await
        .unwrap();

    tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;


}

Searchable metadata is created from the Book details, and the isbn13 is used as the object key. I added the sleep statement, so I didn't reach AWS rate limits when sending items.

Once items were uploaded, I validated the process using s3vectors-embed tool

s3vectors-embed query \
  --vector-bucket-name s3-vector-books \
  --index-name books \   
  --model-id amazon.titan-embed-text-v2:0 \
  --query-input "space exploration" \
  --k 10S

Query data with Lambda

The lambda handler is pretty simple

use aws_config::BehaviorVersion;
use common::{embeddings_svc::EmbeddingsService, s3Vector_svc::S3VectorsService};
use lambda_http::{run, service_fn, Error, IntoResponse, Request, RequestPayloadExt, Response};

use serde::Deserialize;

#[derive(Deserialize)]
struct Query {
    query: String,
    category: Option<String>
}

async fn function_handler(embeddings_svc: &EmbeddingsService, s3_vector_svc: &S3VectorsService, event: Request) -> Result<impl IntoResponse, Error> {

    let query = event.payload::<Query>()?.unwrap();

    let embeddings = embeddings_svc.get_embeddings(query.query.clone()).await;

    let books = s3_vector_svc.query_books(embeddings.embedding, query.category).await;


    let resp = Response::builder()
        .status(200)
        .header("content-type", "application/json")
        .body(serde_json::to_string(&books).unwrap())
        .map_err(Box::new)?;

    Ok(resp)
}

#[tokio::main]
async fn main() -> Result<(), Error> {
    tracing_subscriber::fmt()
        .with_max_level(tracing::Level::INFO)
        // disable printing the name of the module in every log line.
        .with_target(false)
        // disabling time is handy because CloudWatch will add the ingestion time.
        .without_time()
        .init();

    let config = aws_config::defaults(BehaviorVersion::latest()).region("us-east-1").load().await;

    let embeddings_svc = EmbeddingsService::new(&config).await;

    let s3_vector_svc = S3VectorsService::new(&config).await;

    run(service_fn(|ev| {
        function_handler(&embeddings_svc, &s3_vector_svc, ev)
    })).await
}

I expect to receive the query and the optional category to filter results. S3 Vectors provide basic metadata filtering functionalities. It might be super useful.

To query S3 I get the embeddings from a Bedrock and I use them to the s3_vector call.

// rust_app/common/src/s3Vector_svc.rs
pub async fn query_books(&self, vector: Vec<f32>, category: Option<String>) -> Vec<Book> {

        let data = VectorData::Float32(vector);

        let category_filter = category
            .map(|category| {
                let mut category_json = HashMap::new();
                category_json.insert("categories".to_string(), Document::String(category));
                Document::Object(category_json)
            });

        let request = self.client
            .query_vectors()
            .vector_bucket_name(S3_NAME)
            .index_name(INDEX_NAME)
            .top_k(5)
            .query_vector(data)
            .set_filter(category_filter)
            .return_metadata(true);

        let result = request
            .send()
            .await
            .unwrap();

        let data = result
            .vectors()
            .iter()
            .map(|vector| {
                let id = vector.key().parse().unwrap();
                let metadata = vector.metadata().unwrap().as_object().unwrap();
                let source_text = metadata.get("source_text").unwrap().as_string().unwrap().to_string();;
                let title = metadata.get("title").unwrap().as_string().unwrap().to_string();
                let authors = metadata.get("authors").unwrap().as_string().unwrap().to_string();
                let categories = metadata.get("categories").unwrap().as_string().unwrap().to_string();

                Book { isbn13: id, title, authors, categories, description: source_text }
            })
            .collect::<Vec<_>>(); 

        println!("Result: {:?}", result);

        data
    }

As you see, there are unwrap() everywhere. I truly believe it is the way to prototype in Rust.
The only interesting thing in this code is that Rust s3_vectorsSDK uses a generic Document type from the Smithy.

For local testing, I use cargo lambda tool. Once it works locally, I deploy it to AWS.

Deploy and test

I used a generic SAM template for the Rut project. The only thing is that it didn't work for me with the workspace in the Rust project. It is strange because I am using AWS CDK in my projects, and I face no issues. Anyway, I will build the project in a separate step and point the handler definition in SAM to the binary.

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: >
  s3_vector

  Sample SAM Template for s3_vector

Globals:
  Function:
    Timeout: 3
    MemorySize: 128


Resources:
  FindBooksFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: ./rust_app/target/lambda/handler
      Handler: bootstrap
      Runtime: provided.al2023
      Architectures:
        - x86_64
      Policies: 
        - Version: "2012-10-17"
          Statement:
            - Effect: Allow
              Action: bedrock:InvokeModel
              Resource: "arn:aws:bedrock:*::foundation-model/amazon.titan-embed-text-v2:0"
        - Version: "2012-10-17"
          Statement:
            - Effect: Allow
              Action: 
                - s3vectors:QueryVectors
                - s3vectors:GetVectors
              Resource: "arn:aws:s3vectors:us-east-1:${AWS::AccountId}:bucket/s3-vector-books/index/books"
      Events:
        GetBooks:
          Type: Api 
          Properties:
            Path: /books
            Method: post

Outputs:
  FindBooksApi:
    Description: "API Gateway endpoint URL for Prod stage for find books"
    Value: !Sub "https://${ServerlessRestApi}.execute-api.${AWS::Region}.amazonaws.com/Prod/books"

I was very curious about how much latency I should expect. Turned out that for the cold start, the Lambda is finishing under one second.

The hot one takes around 300ms. Adding category doesn't really impact the query.

To be honest, those numbers are quite amazing for this kind of serverless workload.

Summary

CODE IS AVAILABLE IN THIS REPO

The most amazing thing about S3 Vector Buckets is that they don't require any computing. You don't need clusters or instances to query vectors. It allows a fully serverless approach.
When combined with AWS Lambda written in Rust, the overall overhead of performing semantic search becomes relatively small. The documentation promises sub-second query performance, and in my case, I was able to complete the whole request under one second when dealing with a cold start.

There are, of course, some trade-offs. Even though the metadata filtering is pretty handy, it doesn't allow more complex queries. If the similarity search needs to be done in the wider context, S3 Vector buckets might not be enough.

Anyway, it is interesting to see what direction this project takes before becoming generally available.

MCP OAuth on AWS Lambda with WorkOS

szymon-szym — Mon, 09 Jun 2025 18:07:39 +0000

Photo by Bruno Alves on Unsplash

Goal

I plan to implement an MCP server secured with OAuth. I will use a third-party identity provider (WorkOS for this blog post, but general rules would be the same for other services). The created resource is meant to be used by users in their MCP clients.

The main idea is to build a serverless solution that might integrate with the existing authentication flow. From the user's perspective, using MCP Server should feel similar to using other services.

Problem

The current MCP specification assumes that the MCP Server will handle authorization as well. It might work if you have a single MCP server, but it doesn't fit well into an enterprise landscape, with dozens of services and a separate layer for identity provider.

Architecture

I will create two separate services behind the API Gateway. One is responsible for the authorization bridge with WorkOS, the other for the actual MCP server logic.

I use a stateless version of the MCP server, so both services will run on Lambda Functions.

API Gateway will be responsible for authorizing the calls to the resource server (using Lambda authorizer). With this approach, my MCP resource server won't know anything about authorization logic.

Implementation

The whole code is available in this repo

Initial setup

To be honest, I don't fully understand the MCP specification, especially when it comes to authorization. Moreover, MCP clients do not rush to implement the updated specification, which makes things even more tricky.

I will test my server with goose ( open source AI client ) using the mcp-remote package for handling a Streamable HTTP type of MCP server.

To begin, I create a simple mock authorizer that returns Unauthorized by default.

// authorizer/src/main.rs
use std::collections::HashMap;

use lambda_runtime::{run, service_fn, tracing, Error, LambdaEvent};
use serde::{Deserialize, Serialize};
mod generic_handler;

// https://docs.aws.amazon.com/apigateway/latest/developerguide/http-api-lambda-authorizer.html

#[derive(Deserialize)]
pub struct Request {
    pub headers: HashMap<String, String>
}

#[derive(Serialize)]
pub struct Response {
    pub isAuthorized: bool,
    pub context: HashMap<String, String>
}


#[tokio::main]
async fn main() -> Result<(), Error> {
    tracing::init_default_subscriber();

    run(service_fn(function_handler)).await
}

async fn function_handler(event: LambdaEvent<Request>) -> Result<Response, Error> {
    // auth logic 
    Ok(Response {
        isAuthorized: false,
        context: HashMap::new()
    })
}

My initial MCP resource server returns a dummy message in response to POST request on the /mcp route

// mcp-resource/src/main.rs
use axum::{http::StatusCode, response::IntoResponse, routing::post, Router};
use tower_http::trace::{DefaultMakeSpan, TraceLayer};
use tracing_subscriber::EnvFilter;

const PORT: u16 = 8080;

#[tokio::main]
async fn main() {

    tracing_subscriber::fmt()
        .with_env_filter(
            EnvFilter::try_from_default_env()
                .unwrap_or_else(
                    |_| EnvFilter::new("debug")
                )
        )
        .init();

    let app = Router::<()>::new()
        .route("/mcp", post(mcp_handler))
        .layer(
            TraceLayer::new_for_http()
                .make_span_with(DefaultMakeSpan::new()
                    .include_headers(true)
                )
        );

    let listener = tokio::net::TcpListener::bind(format!("0.0.0.0:{}", PORT)).await.unwrap();

    axum::serve(listener, app).await.unwrap();
}

pub async fn mcp_handler() -> impl IntoResponse {
    "hello"
}

Finally, the infrastructure defines a HTTP API Gateway and both functions. My MCP resource server runs the axum application behind Lambda Web Adapter

infrastructure/lib/infrastructure-stack.ts
import * as cdk from 'aws-cdk-lib';
import { Construct } from 'constructs';
import { RustFunction } from '@cdklabs/aws-lambda-rust'

export class MCPAuthLambda extends cdk.Stack {
  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props);

    const authorizer = new RustFunction(this, 'Authorizer', {
      entry: '../authorizer/Cargo.toml',
      environment: {
        RUST_LOG: 'info'
      }
    })

    const adapterLayer = cdk.aws_lambda.LayerVersion.fromLayerVersionArn(
      this,
      'WebAdapterLayer',
      'arn:aws:lambda:us-east-1:753240598075:layer:LambdaAdapterLayerX86:25'
    );

    const mcpResourceServer = new RustFunction(this, 'McpResourceServer', {
      entry: '../mcp-resource',
      layers: [ adapterLayer ],
      environment: {
        RUST_LOG: 'info'
      }
    })

    const httpApi = new cdk.aws_apigatewayv2.HttpApi(this, 'MCP-Auth-API');

    httpApi.addRoutes({
      path: '/mcp',
      authorizer: new cdk.aws_apigatewayv2_authorizers.HttpLambdaAuthorizer(
        'Authorizer', authorizer, {
        responseTypes: [cdk.aws_apigatewayv2_authorizers.HttpLambdaResponseType.SIMPLE]
      }),
      methods: [cdk.aws_apigatewayv2.HttpMethod.GET, cdk.aws_apigatewayv2.HttpMethod.POST],
      integration: new cdk.aws_apigatewayv2_integrations.HttpLambdaIntegration(
        "mcp-resource",
        mcpResourceServer)
    });

    new cdk.CfnOutput(
      this,
      'ApiUrl',
      {
        value: httpApi.url!
      }
    )

  }
}

Local testing

I use SAM local to run APIs locally and test. To be able to do so with AWS CDK, I run cdk synth in the infrastructure directory, and then sam local start-api -t MCPAuthLambda.template.json in the cdk.out directory.

Now I will add my MCP to the goose by starting

goose configure

and adding mcp-remote extension:

Now, when I start goose, it tries to use my MCP server:

Ok, as expected, the goose tries to connect to the /mcp route, and receives the 401 response. Then it tries to get /.well-known/oauth-authorization-server and fails.

The next step would be to start implementing our authorization logic.

Authorization Server

`/.well-known/oauth-authorization-server`

According to the specification, the first step in the auth process is metadata discovery. In other words, the MCP client queries for the information about expected endpoints, and falls back to the default ones if the .well-known endpoint is not available.

Let's start from the well-known endpoint

// mcp-authorize/src/main.rs
use axum::{debug_handler, response::IntoResponse, routing::get, Json, Router};

const PORT: u16 = 8081;

#[tokio::main]
async fn main() {

    let app = Router::<()>::new().route(
        "/.well-known/oauth-authorization-server",
        get(well_known_handler),
    );

    let listener = tokio::net::TcpListener::bind(format!("0.0.0.0:{}", PORT))
        .await
        .unwrap();


    axum::serve(listener, app).await.unwrap();
}

#[debug_handler]
async fn well_known_handler() -> impl IntoResponse {

        Json(WellKnownAnswer {
            authorization_endpoint: "http://localhost:3000/authorize".to_string(),
            registration_endpoint: "http://localhost:3000/register".to_string(),
            grant_types_supported: vec!["authorization_code".to_string(), "refresh_token".to_string()],
            scopes_supported: vec!["email".to_string(), "offline_access".to_string(), "openid".to_string(), "profile".to_string()],
            response_modes_supported: vec!["query".to_string()],
            response_types_supported: vec!["code".to_string()],
            token_endpoint: "http://localhost:3000/token".to_string(),
            issuer: "https://fresh-lamb-82-staging.authkit.app".to_string(),
        })

}


#[derive(serde::Serialize)]
pub struct WellKnownAnswer {
    pub authorization_endpoint: String,
    pub registration_endpoint: String,
    pub grant_types_supported: Vec<String>,
    pub scopes_supported: Vec<String>,
    pub response_modes_supported: Vec<String>,
    pub response_types_supported: Vec<String>,
    pub token_endpoint: String,
    pub issuer: String,
}

I also updated the infrastructure with the newly created service

// ...
const mcpAuthorizationServer = new RustFunction(this, 'McpAuthorizationServer', {
      entry: '../mcp-authorize',
      layers: [ adapterLayer ],
      environment: {
        RUST_LOG: 'debug',
        PORT: '8081',
      }
    })
//...
httpApi.addRoutes({
      path: '/.well-known/oauth-authorization-server',
      methods: [cdk.aws_apigatewayv2.HttpMethod.GET],
      integration: new cdk.aws_apigatewayv2_integrations.HttpLambdaIntegration(
        "mcp-authorization",
        mcpAuthorizationServer)
    });
// ...

Rebuild the project with cdk synth and run sam local

Now, when starting goose, I can see that we are one step further

`/register`

MCP spec expects that the OAuth flow allows dynamic client registration. In my case, I don't plan to register the new client_id each time, and I will use /register endpoint to plug the configured WorkOS server.

I take the client_id from the main page of the WorkOS dashboard, https://dashboard.workos.com/get-started

According to the RFC 7591, the client_id is the only required field in the response.
It turned out that mcp-remote also requires the redirect_uris property in the response. I will take it from the payload of the request sent to the /register endpoint

// mcp-authorize/src/main.rs
// ...

    let app = Router::<()>::new().route(
        "/.well-known/oauth-authorization-server",
        get(well_known_handler),
    )
    .route("/register", post(registration_handler));
// ...
#[debug_handler]
async fn registration_handler(Json(req): Json<ClientRegistrationRequest>) -> impl IntoResponse {
    println!("{:?}", req);
    Json(ClientRegistrationAnswer{
        client_id: CLIENT_ID.to_string(),
        redirect_uris: req.redirect_uris,
        response_types: req.response_types,
        grant_types: req.grant_types,
        client_name: req.client_name
    })
}

#[derive(Debug, Deserialize)]
pub struct ClientRegistrationRequest {
    pub client_name: String,
    pub redirect_uris: Vec<String>,
    pub grant_types: Vec<String>,
    pub token_endpoint_auth_method: String,
    pub response_types: Vec<String>,
}

#[derive(Serialize)]
pub struct ClientRegistrationAnswer {
    client_id: String,
    redirect_uris: Vec<String>,
    response_types: Vec<String>,
    grant_types: Vec<String>,
    client_name: String
}
//...

I add the endpoint to APIs

// ...
    httpApi.addRoutes({
      path: '/register',
      methods: [cdk.aws_apigatewayv2.HttpMethod.POST],
      integration: new cdk.aws_apigatewayv2_integrations.HttpLambdaIntegration(
        "mcp-authorization",
        mcpAuthorizationServer)
    });

And start goose one more time

Now the default browser is opened with the call for the /authorize endpoint

`/authorize`

When performing authorization, my goal is to return redirect to the user with the proper parameters, so WorkOS will perform heavy lifting.
I built the logic based on the documentation.
The only tricky part is that I can't simply use the redirect_uri sent by the MCP client, as I have not added it to the list of allowed URIs in the WorkOS Dashboard. It happens because we are not registering the client dynamically, so our underlying authorization server doesn't know about the specific redirect_uri sent with the request.
I solve it by storing the original redirect uri as a state, and using the URI registered on my server:

// ...
#[debug_handler]
async fn authorization_handler(Query(params): Query<HashMap<String, String>>) -> impl IntoResponse {

    let original_uri = params.get("redirect_uri").unwrap();

    let state = BASE64_STANDARD.encode(original_uri);

    let local_redirect = "http://localhost:3000/callback";

    let url = reqwest::Url::parse_with_params(
        "https://api.workos.com/user_management/authorize",
        &[
            ("response_type", "code"),
            ("client_id", CLIENT_ID),
            ("redirect_uri", local_redirect),
            ("code_challenge", params.get("code_challenge").unwrap()),
            ("code_challenge_method", "S256"),
            ("provider", "authkit"),
            ("state", state.as_str()),
            ("scope", "openid profile email offline_access"),
        ],
    )
    .unwrap();

    Redirect::temporary(url.as_str())

}
// ...

BTW - when prototyping, I use some hardcoded variables, and I will clean them up later, when deploying to AWS.

Now I start goose again and see a login page open in the default browser:

It is super convenient to have the whole UI implemented by WorkOS AuthKit.

`/callback`

This endpoint is not part of the specification, but it is needed for statically created clients with registered redirect_uris. I will get the response from the authorize action and redirect it to the original redirect_uri stored in the request state

// ...
#[debug_handler]
async fn callback_handler(Query(params): Query<HashMap<String, String>>) -> impl IntoResponse {
    let code = params.get("code").unwrap();

    let state = params.get("state").unwrap();

    let original_redirect_uri = String::from_utf8(BASE64_STANDARD.decode(state).unwrap()).unwrap();

    let response_url = reqwest::Url::parse_with_params(
        original_redirect_uri.as_str(),
        &[("code", &code), ("state", &state)],
    )
    .unwrap();

    Redirect::temporary(response_url.as_str())
}
// ...

Now, after logging in, I can see that the code was properly redirected

`/token`

The final part of the authorization process is to exchange the code for the token.

Our flow allows two types of flows authorization_code and redresh_token. Token handler checks the type of the request, and goes to the WorkOS to get the actual token.

async fn token_handler(State(state): State<AppState>, Form(form): Form<HashMap<String, String>>) -> impl IntoResponse {

    let grant_type = form.get("grant_type").unwrap();

    let client = reqwest::Client::new();

    let request = match grant_type.as_str() {
        "authorization_code" => {
            let code = form.get("code").unwrap();
            let code_verifier = form.get("code_verifier").unwrap();
            let client_id = form.get("client_id").unwrap();

            json!({
                "client_id": client_id,
                "client_secret": state.workos_client_secret,
                "grant_type": "authorization_code",
                "code": code,
                "code_verifier": code_verifier
            })
        }
        "refresh_token" => {
            let refresh_token = form.get("refresh_token").unwrap();
            let client_id = form.get("client_id").unwrap();

            json!({
                "client_id": client_id,
                "client_secret": state.workos_client_secret,
                "grant_type": "refresh_token",
                "refresh_token": refresh_token,
            })
        }
        _ => panic!("Invalid grant type"),
    };

    println!("token request: {}", request);

    let res = client
        .post("https://api.workos.com/user_management/authenticate")
        .body(serde_json::to_string(&request).unwrap())
        .header("Content-Type", "application/json")
        .send()
        .await
        .unwrap();

    let authkit_response: AuthkitAuthResult =
        serde_json::from_str(res.text().await.unwrap().as_str()).unwrap();

    println!("token response {:?}", &authkit_response);

    let expires_at = chrono::Utc::now().timestamp() + 3600;

    let token_result = TokenResponse {
        access_token: authkit_response.access_token,
        refresh_token: authkit_response.refresh_token,
        token_type: "Bearer".to_string(),
        expires_at,
    };

    Json(token_result)
}
//...
#[derive(Serialize, Deserialize, Debug)]
pub struct TokenResponse {
    access_token: String,
    refresh_token: String,
    token_type: String,
    expires_at: i64,
}

#[derive(Serialize, Deserialize, Debug)]
pub struct AuthkitAuthResult {
    access_token: String,
    refresh_token: String,
}
// ...

Quick check and it looks that the authorization flow is successfully finished, as now I can see the error from the initial /mcp endpoint - I need to implement API Gateway authorizer

Authorizer

I've planned to set up everything and test locally before even deploying to AWS. It turned out that it is not possible to fully test the lambda authorizer with sam local start-api command.

Lambda authorizer is quite straightforward. I am getting jwks and decode JWT token.

// authorizer/src/token.rs

use jsonwebtoken::{decode_header, DecodingKey, TokenData};
use serde::{Deserialize, Serialize};

#[derive(Debug, Serialize, Deserialize)]
pub struct Claims {
    pub sub: String,
    pub exp: usize,
}

#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct JWK {
    pub kid: String,
    pub kty: String,
    pub alg: String,
    pub n: String,
    pub e: String,
}

#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct JWKS {
    pub keys: Vec<JWK>,
}

pub async fn get_jwks(jwks_url: String) -> JWKS {
    let jwks_resp = reqwest::get(jwks_url).await.unwrap();

    let jwks: JWKS = jwks_resp.json().await.unwrap();

    jwks

}

pub async fn check_token(token: &str, keys: &JWKS) -> Result<TokenData<Claims>, String> {

    let header = decode_header(token).unwrap();

    let kid = header.kid.ok_or("No kid found in token header")?;

    let jwk = keys.keys.iter()
        .find(|k| k.kid == kid)
        .ok_or("No matching kid found in jwks")?;

    let decoding_key = DecodingKey::from_rsa_components(&jwk.n, &jwk.e)
        .map_err(|op| format!("Error: {:?}", op))?;

    let mut validation = jsonwebtoken::Validation::new(jsonwebtoken::Algorithm::RS256);

    validation.validate_exp;

    let token_data = jsonwebtoken::decode::<Claims>(token, &decoding_key, &validation)
    .map_err(|op| format!("Error: {:?}", op))?;

    println!("{:?}", token_data.claims);

    Ok(token_data)
}

And the handler

// authorizer/src/main.rs

use std::collections::HashMap;

use lambda_runtime::{run, service_fn, tracing, Error, LambdaEvent};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use token::JWKS;

mod token;

// https://docs.aws.amazon.com/apigateway/latest/developerguide/http-api-lambda-authorizer.html

#[derive(Deserialize)]
pub struct Request {
    pub headers: HashMap<String, String>,
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct Response {
    pub is_authorized: bool,
    pub context: HashMap<String, String>,
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct UnauthorizedResponse {
    pub error_message: String,
}

#[derive(Serialize)]
#[serde(untagged)]
pub enum AuthorizerResponse {
    PermissionResponse(Response),
    Unauthorized(UnauthorizedResponse),
}

#[tokio::main]
async fn main() -> Result<(), Error> {
    tracing::init_default_subscriber();

    let jwks_url = std::env::var("JWKS_URL").expect("JWKS_URL must be set");

    let jwks = token::get_jwks(jwks_url).await;

    println!("JWKS: {:?}", &jwks);

    run(service_fn(|ev| function_handler(&jwks, ev))).await
}

async fn function_handler(
    jwks: &JWKS,
    event: LambdaEvent<Request>,
) -> Result<AuthorizerResponse, Error> {
    let token_header = event.payload.headers.get("authorization");

    if token_header.is_none() {
        println!("No token header");
        return Ok(AuthorizerResponse::Unauthorized({
            UnauthorizedResponse {
                error_message: "Unauthorized".to_string(),
            }
        }));
    }

    let token = token_header.unwrap().replace("Bearer ", "");

    let claims = token::check_token(token.as_str(), jwks).await;
    // auth logic

    match claims {
        Ok(tk) => {
            println!("Claims: {:?}", tk);
            Ok(AuthorizerResponse::PermissionResponse(Response {
                is_authorized: true,
                context: HashMap::new(),
            }))
        }
        Err(e) => {
            println!("Error: {:?}", e);
            Ok(AuthorizerResponse::Unauthorized({
                UnauthorizedResponse {
                    error_message: "Unauthorized".to_string(),
                }
            }))
        }
    }
}

The only interesting part is that when I return isAuthorized as false, it is translated by API Gateway to a 403 status. It won't work with MCP flow, which requires a 401 status to trigger authorization discovery.

I have moved hardcoded things to env variables and deployed the project to AWS.

MCP Server

The last missing part is the actual MCP Server

For simplicity's sake, I take the Counter example from the examples for official Rust MCP SDK

Test

To test the whole solution, I add the new extension to goose and start the program

I can see the login window

After logging in, I am properly redirected to the localhost used by goose

Now I can use my MCP counter service that runs on Lambda behind AWS API Gateway. The connection is secured by the token obtained by the client from the WorkOS server

The whole code is available in this repo

Summary

In this blog post, I implemented the authorization service for the MCP Server using WorkOS as an identity provider. This way, clients using my server in their MCP clients would be able to log in using a third-party identity provider. This is a common flow in organizations that probably already have this provider set up.

At the moment, MCP clients don't fully implement the updated MCP schema, so even using Streamable HTTP type of servers requires some kind of workarounds, not to speak of authorization flow. I believe that with time, the ecosystem will mature, and providing clients with secure MCP servers will become more straightforward.

Plan school lunches for kids with Amazon Q

szymon-szym — Tue, 06 May 2025 19:20:35 +0000

This is a submission for the Amazon Q Developer "Quack The Code" Challenge: Exploring the Possibilities

Photo by Eaters Collective on Unsplash

What I Built

Problem to solve - "I don't like it"

Making sure that my kids will eat the lunch

In my sons' school, there are two main lunch options - you can order a meal from the catering provider, or bring your own food. Each option has its own pros and cons, but the main thing is that I cannot order lunch from the catering service whenever I want to. There is a specific menu, and kids, like everyone else, have their food preferences. Someone likes meat in all forms, someone won't touch a traditional Polish port chop

Currently, in my family, the process of "providing our kids with lunch" is more or less spontaneous:

what is tomorrow's lunch?
they won't eat this ...
we don't have products to prepare lunch, I need to go for groceries ...

and so far, and so on

The solution I prepared is aimed at helping us have more control over planning meals for our kids.

Solution

As a result, I would like to create a page in my Notion to easily track the menu for upcoming days. I will also let q suggest some recipes for meals prepared at home.

Step 1 - Get menu

So catering service provides a well-documented API to use when getting menu ... Just kidding! Who would build APIs for the catering service?

To get the menu, I need to log in to the webpage.

q can do this. There is a reason why the Playwright MCP Server is probably the most popular tool in the MCP ecosystem. With Playwright, AI agents can go and actually use the Internet.

The whole "code" is available in this GitHub repo (there is no code, just configuration and prompts)

I create mpc.json file and add playwright tool there:

.amazonq/mcp.json

{
    "mcpServers": {
        "playwright": {
            "command": "npx",
            "args": [
                "@playwright/mcp@latest"
            ]
        }
    }
}

Now, when I run q, I can ask it to go to a specific page:

And after a few actions I see the video with funny cats in the browser.

In the same way, I will scrape the menu from the website using the same playwright tool.

Let's start from the first prompt in the get_menu.md

- read catering/.env file and get CATERING_LOGIN and CATERING_PASSWORD
- go to https://www.zamow.kuchniacateringowa.pl/ and log in
- go to https://www.zamow.kuchniacateringowa.pl/Product/Menu
- click "Pokaż menu" for "SZKOŁA PODSTAWSOWA- KLASY 1-8"
- check the "II danie + zupa" and "II danie + zupa + deser" for each date and create a table in the markdown format with the following columns
    - date
    - main dish (keep it in Polish)
    - desert (keep it in Polish)
- save the table as a <last_date_from_menu>.md in the catering/menu/ directory

Now I ask q to read the instructions and perform actions

For some reason, sometimes the last step is not performed. I need to ask again to save the file

As a result menu is saved in the new file.

Step 2 - Analyze preferences

I decided to split the whole process into steps so I could easily rerun the specific part without going through all the steps again. Once I have the menu saved, I can analyze food preferences and iterate over different inputs.

The next prompt in the analyze_menu.mg

- go to catering/menu directory and read the file with the newest date in the title. the date needs to be not earlier than tomorrow, otherwise stop the process
- read preferences from catering/preferences directory 
- check the "Main Dish" column from the menu file for each date and create a table in the markdown format with the following columns
    - date
    - main dish (keep it in Polish)
    - desert (keep it in Polish)
    - OK for dummy_name_1 (use emoji for OK, NOT, SO-SO)
    - comment dummy_name_1 (based on loaded preferences, why this dish will be OK, NOT or SO-SO)
    - OK for dummy_name_2 (use emoji for OK, NOT, SO-SO)
    - comment dummy_name_2 (based on loaded preferences, why this dish will be OK, NOT or SO-SO)
- save the table as a <last_date_from_table>.md in the catering/analyze/ directory

I created files with preferences for each of my sons. You might wonder why they are called Dummy_name_1 and Dummy_name_2, but hey, naming part is the hardest one, right?

Now let's ask q to do the magic

I save the result as a markdown file, and I can review it locally.

🎉 🎉

It looks amazing already! Let's move forward

Step 3 - Upload the result to Notion

I use Notion for planning already, so it would be nice to have a result table there. Now I can check the menu with my checks anytime on my phone. And it looks nice.

Prepare Notion page

I have created a parent page in the Notion and saved its ID.

I also created an internal integration to be able to use its token

Finally, I added integration to my target page

To interact with notion I will use the official notion MCP. I need to update my mcp.json file:

{
    "mcpServers": {
        "playwright": {
            "command": "npx",
            "args": [
                "@playwright/mcp@latest"
            ]
        },
        "notionApi": {
          "command": "npx",
          "args": ["-y", "@notionhq/notion-mcp-server"],
          "env": {
            "OPENAPI_MCP_HEADERS": "{\"Authorization\": \"Bearer <YOUR_NOTION_TOKEN>", \"Notion-Version\": \"2022-06-28\" }"
          }
        }
    }
}

It's time to run the q with the following prompt from the update_notion.md

- read the file with the latest date in the title from the catering/analyze/ directory
- post page with the content from the file named as a file. parent page ID is 1eac5262835580c28f59cd9bd1e24ae1
- use Notion formatting to create a page to preserve headings and to add a table
- page should be named "Catering analysis <date_from_the_file_name>"
- if there is already a page with this name as a child of 1eac5262835580c28f59cd9bd1e24ae1, update the existing page instead of creating the new one

And there is a notion page created

How cool is this!

I can share this page with my wife, so we both have visibility. Or with my kids and ask them to validate results, so I can improve the definition of their preferences

Step 4 - Suggest recipes

Once I know when I would probably need to prepare lunch at home, I could use some advice on the ideas for recipes.

Prompt from prepare_recipes.md

- read the file with the latest date in the title from the catering/analyze/ directory
- look for dates where the dish is not ok for at least one kid
- read preferences from catering/preferences directory 
- for those days suggest recipes for dishes easy to prepare at home the day before and easy to warm up. If there are two days in the row, the recipe could use the same "base" to prepare lunches for two days
- even if there are recipes for two consecutive days, show both as a separate row in the table
- recipes should be simple and fast, e.g. without preparing dumpings, as it is time consuming
- recipes should include all macro elements needed, vegetables and meat, or other source od protein
- avoid sugar 
-  save recipes as a markdown table with date, recipe and ingredients. the file name should be menu/<date_from_the_analyze_file>_recipes.md

And the result

Not bad!

Summary

I used q to solve a real problem and to make my life a bit easier. Interacting with q via the console was fun. I really liked the ease of integrating MCP-based tools.

Not to say that the possibility to use q with the BuilderID for free is a super nice option!

q deals with non-English content without any problem.

Next steps

In the future, I would like to gather information about the current offers from the shops where I used to buy groceries. I am going to use this info when planning meals.

Maybe my q agent should just order some food for me, or at least put items in the basket in the online checkout?

Code Repository

https://github.com/szymon-szym/amazonq-plan-lunch

How I Used Amazon Q Developer

built-in tools to read files with prompts, preferences, and .env with configuration and credentials
Playwright MCP to use a headless browser to access the Internet
Notion MCP to update the page on Notion
LLMs behind q to reason about food preferences and to suggest recipes

MCP Server with AWS Lambda and HTTP Api Gateway

szymon-szym — Sun, 04 May 2025 17:37:51 +0000

Photo by Mika Baumeister on Unsplash

For a "long" time (long in the AI terms) the serverless approach didn't really fit the Model Context Protocol idea. MCP was designed with statefulness in mind, and the ephemeral computing didn't match the initial requirements.

The situation changed with the latest protocol update, which replaced SSE + HTTP transport with Streamable HTTP transport. This is a significant update, which opened up new possibilities. Now MCP server can be stateless or can handle state using sessionId from the request header. It means that using a serverless approach doesn't require any tricks from our side.

Goal

I build an MCP Server and deploy it as the AWS Lambda function behind HTTP Gateway. For the moment I am writing the blog post, the Streamable HTTP transport is available out-of-the-box only in the TypeScript sdk. For other languages, there is already work in progress for implementing the latest changes, so there is a huge chance that when you read this, you can pick another SDK.

I create an MCP Server that acts as a prompt library. It will use S3 as a storage for prompt templates, and allow dynamic prompt creation.

Architecture

I use AWS CDK to define infrastructure.

Code for this blog post is available HERE

Server function

The TypeScript SDK for MCP has great documentation, including examples. I will use one of them as a starting point.

After creating a directory and installing dependencies, I create a function for the MCP server initialization

// server/server.ts
import { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js'
import { GetPromptResult } from '@modelcontextprotocol/sdk/types';
import { z } from 'zod';

export const initializeServer = (): McpServer => {
    // Create an MCP server with implementation details
    const server = new McpServer({
      name: 'prompt-gallery-server',
      version: '1.0.0',
    }, { capabilities: { logging: {} } });

     // Register a simple prompt
    server.prompt(
        'greeting-template',
        'A simple greeting prompt template',
        {
        name: z.string().describe('Name to include in greeting'),
        },
        async ({ name }): Promise<GetPromptResult> => {
        return {
            messages: [
            {
                role: 'user',
                content: {
                type: 'text',
                text: `Please greet ${name} in a friendly manner.`,
                },
            },
            ],
        };
        }
    );

    return server
}

And I use it for handling requests in the express app

// main.ts
import express, { Request, Response } from "express";
import { initializeServer } from "./server/server";
import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";

const app = express();
app.use(express.json());

app.post("/mcp", async (req, res) => {
  const server = initializeServer();
  try {
    const transport = new StreamableHTTPServerTransport({
      sessionIdGenerator: undefined,
    });

    await server.connect(transport);

    await transport.handleRequest(req, res, req.body);

    req.on("close", () => {
      console.log("Request closed");
      transport.close();
      server.close();
    });
  } catch (e) {
    console.error("Error handling MCP request:", e);
    if (!res.headersSent) {
      res.status(500).json({
        jsonrpc: "2.0",
        error: {
          code: -32603,
          message: "Internal server error",
        },
        id: null,
      });
    }
  }
});

app.get('/mcp', async (req: Request, res: Response) => {
    console.log('Received GET MCP request');
    res.writeHead(405).end(JSON.stringify({
      jsonrpc: "2.0",
      error: {
        code: -32000,
        message: "Method not allowed."
      },
      id: null
    }));
  });

  app.delete('/mcp', async (req: Request, res: Response) => {
    console.log('Received DELETE MCP request');
    res.writeHead(405).end(JSON.stringify({
      jsonrpc: "2.0",
      error: {
        code: -32000,
        message: "Method not allowed."
      },
      id: null
    }));
  });

  const PORT = 3000;

  app.listen(PORT, () => {
    console.log(`MCP server running on ${PORT}`)
  })

Ok, now let's run

npx ts-node main.ts

Local testing

Once it is running, I want to test it. The most convenient way is to use MCP Inspector, which will test our server against specific rules expected by the protocol.

When you read this blog post, you can probably run inspector directly using npx, but at this day I need to clone the repo and run npm run dev to be able to pick the HTTP Streamable connection

Great, I can connect to my server, and I see that it offers prompts. I can list available prompts, and call the only one available with the name parameter

Cool. So far, so good!

Cloud infrastructure

We need some extra steps before deploying the Express app as a Lambda function. AWS Lambda has its own mechanism to pass requests to the function, as it can be called in many different ways. Luckily, there is an awesome AWS Lambda web adapter that allows deploying web applications as a lambda function and provides a proxy to translate the request from HTTP/REST Gateway to the shape expected by a web application.

Lambda Web Adapter is an extension written in Rust, which is added to the function as a layer. In the infrastructure definition lambda function is set as a default integration in the Http Api

// ...
    const mcpHttpHandler = new cdk.aws_lambda_nodejs.NodejsFunction(
      this,
      "httpHandler",
      {
        entry: "../server/main.ts",
        runtime: cdk.aws_lambda.Runtime.NODEJS_20_X,
        timeout: cdk.Duration.minutes(3),
        handler: "run.sh",
        architecture: cdk.aws_lambda.Architecture.X86_64,
        environment: {
          AWS_LAMBDA_EXEC_WRAPPER: "/opt/bootstrap",
          RUST_LOG: "info",
        },
        bundling: {
          minify: true,
          commandHooks: {
            beforeInstall: () => [],
            beforeBundling: () => [],
            afterBundling: (inputDir: string, outputDir: string) => {
              return [`cp ${inputDir}/../server/run.sh ${outputDir}`];
            },
          },
          target: "node20",
          externalModules: ["@aws-sdk/*"],
          sourceMap: true,
          forceDockerBundling: false,
        },
        layers: [
          LayerVersion.fromLayerVersionArn(
            this,
            "layer",
            `arn:aws:lambda:us-east-1:753240598075:layer:LambdaAdapterLayerX86:25`
          ),
        ],
      }
    );

    const mcpHttpApi = new cdk.aws_apigatewayv2.HttpApi(this, "mcpAPI", {
      defaultIntegration: new cdk.aws_apigatewayv2_integrations.HttpLambdaIntegration(
        "httpLambdaIntegration",
        mcpHttpHandler
      ),
    })
// ...

After deployment, I can call the MCP server running as a Lambda Function. Please see the URL used in the Inspector

Authorization

Security is one of the critical aspects that needs to be covered when the MCP matures. We will see how MCP clients will handle authentication in the future. I expect that it will look similar to the standard application flow that allows usage of the external providers, like cognito, okta, etc.

I secure my MCP server with Cognito JWT authorizer.

The thing is that currently, MCP Inspector doesn't fully support authorization for HTTP Streamable transport. At least I wasn't able to have it work, but there is active development in the repository in this area.

To test functionality, I use Bruno and call the server directly.

Without the authorization header

Connect with the token

Get the list of prompts

And finally, generate the prompt

Integrate with S3

Having an MCP server running as the AWS Lambda opens unlimited possibilities to integrate with the serverless infrastructure (and with any other services).

In my case, I will get the template of the prompt from the S3 bucket. Let's add an S3 connection to the function.

Web Applications behind Lambda Web Adapter are similar to the regular Lambdas - we want to initialize all services before the handler, so the time needed for this impacts only the cold start. To achieve it, it is enough to initialize services outside express handlers and inject them into the created server

// initialize S3 service
const client = new S3Client();
const bucketName = process.env.BUCKET_NAME || "bucket name missing";

const s3Service = new S3Service(client, bucketName)

const app = express();
app.use(express.json());

app.post("/mcp", async (req, res) => {
    console.log(`req body: ${JSON.stringify(req.body)}`);

    const server = initializeServer(s3Service);

// ...

Now Lambda Function gets the prompt template from S3.

The full code is available HERE

Summary

Model Context Protocol is evolving pretty fast. The recent changes allowing a stateless approach for creating MCP servers are an example of how rapidly the ecosystem changes.

Model Context Protocol comes with a set of SDKs for different languages. In the Lambda functions, they can be used out-of-the-box for MCP server creation when put behind the Lambda Web Adapter.

In this Blog post, I created a simple stateless server, but thanks to the sessionId passed with the request by the MCP Client, state can be persisted e.g., in the DynamoDB. It allows more complex interactions without giving up the flexibility of the serverless setup.

I believe that opening the MCP specification on the stateless servers will have a significant impact on the whole MCP ecosystem.

Getting structured output from OpenAI with AWS Lambda

szymon-szym — Sun, 30 Mar 2025 17:25:50 +0000

Photo by Hannes Richter on Unsplash

You can ensure that LLM returns the specific JSON shape. This way, your output is typesafe and can be used by other systems.

Goal

Let's say I have a small service that creates screenplays for YouTube videos. I need to ensure that every screenplay has a consistent shape. I want to generate PDFs based on LLM output, and I need it to fit a specific template.

Problem

LLMs, by definition, are not the best in following the specific schema, as they generate responses token by token. Still, different AI providers offer different ways to force the exact response shape.
In this blog post, I use OpenAI API, which allows the response_format property when calling newer models.

Architecture

I am building the lambda function in Rust.

The code is available IN THIS REPO

For interaction with OpenAI API, I use rig.rs, which is a lightweight yet powerful tool for building agentic workflows in Rust. I use only basic features to parse requests and responses to OpenAI API easily.
Creating JSON Schema directly from structs is covered by a wonderful schemars crate.
Deserialization is handled with serde and serde_json

Lambda - initial draft

Service

For now, I don't use models at all. I create a simple service to call OpenAI with a passed prompt without any additional settings.

//screenplay/src/screenplay/service.rs
use rig::{agent::Agent, completion::Prompt, providers::{self, openai::CompletionModel}};

pub struct ScreenplayService {
    agent: Agent<CompletionModel>,
}

impl ScreenplayService {
    pub fn new() -> Self {
        let client = providers::openai::Client::from_env();
        let agent = client
            .agent("gpt-4o")
            .build();

        Self { agent }
    }

    pub async fn generate_screenplay(&self, prompt: &str) -> String {

        let response = self.agent
        .prompt(prompt)
        .await
        .unwrap();

        response
    }
}

Handler

In the root folder, I run cargo lambda new and choose the option for the HTTP handler. The current template contains two files: main and handler, but I used to merge them into a single file. Cargo lambda doesn't update the workspace automatically, so Cargo.toml in the root file needs to be updated manually.

The handler is straightforward - I create a screenplay service and use it in the lambda handler.

//screenplay_handler/src/main.rs
use lambda_http::{run, service_fn, tracing, Body, Error, Request, RequestExt, Response};
use screenplay::screenplay::service::ScreenplayService;

#[tokio::main]
async fn main() -> Result<(), Error> {
    tracing::init_default_subscriber();

    let agent = ScreenplayService::new();
    run(service_fn(|ev| {
        function_handler(&agent, ev)
    })).await

}

pub(crate) async fn function_handler(service: &ScreenplayService, event: Request) -> Result<Response<Body>, Error> {
    // Extract some useful information from the request
    let prompt = event
        .query_string_parameters_ref()
        .and_then(|params| params.first("prompt"))
        .ok_or("expected prompt paramter")?;

    let response = service
        .generate_screenplay(prompt)
        .await;


    let resp = Response::builder()
        .status(200)
        .header("content-type", "text/html")
        .body(response.into())
        .map_err(Box::new)?;
    Ok(resp)
}

With cargo lambda, it is seamless to run lambda locally and check if things are properly lined up. I also create a .env file with my OPENAI_API_KEY

For the test input, I use the template for aws lambda events repo and update rawQueryString "rawQueryString": "prompt=bermuda%20triangle",

I use two terminals. In the first I run cargo lambda watch --env-file .env. Once it runs, I can invoke my lambda in the second terminal cargo lambda invoke -F screenplay_handler/test/input.json

The response is generic, as expected

Context and response schema

Now, the fun part.

OpenAI API allows adding the response_format parameter to the call. It is not part of common AI APIs, so rig.rs doesn't have it as a built-in method, but it is easy to add additional parameters to the call with additional_parameters.

The shape of the request is described in the OpenAI documentation. Long story short, we are expected to pass a json schema.

Here is a simple helper function to create an expected JSON value for rig.rs

// screenplay/src/screenplay/utils.rs
pub fn create_response_format(schema: &Schema, schema_name: &str) -> Value {
    json!({
        "response_format": {
            "type": "json_schema",
            "json_schema": {
                "name": schema_name,
                "strict": true,
                "schema": schema
            }
        }
    })
}

I want to create JSON Schema directly from domain models to make sure that the whole integration is typesafe. I use the shemars crate to seamlessly create a schema from my structs

// screenplay/src/screenplay/models.rs
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize, JsonSchema)]
pub struct Screenplay {
    pub title: String,
    pub scenes: Vec<Scene>,
}

#[derive(Serialize, Deserialize, JsonSchema)]
pub enum Scene {
    Studio {
        location: String,
        time: String,
        description: Option<String>,
        dialogue: Vec<Dialogue>,
    },
    CutTo {
        cut_to: String,
        voiceover: Voiceover,
    },
    FadeOut {
        fade_out: bool,
        screen_text: Option<String>,
        music: Option<String>,
    }
}

#[derive(Serialize, Deserialize, JsonSchema)]
pub struct Dialogue {
    pub character: String,
    pub tone: Option<String>,
    pub lines: String,
}

#[derive(Serialize, Deserialize, JsonSchema)]
pub struct Voiceover {
    pub character: String,
    pub lines: String,
}

Great, now I update the ScreenplayService:

// screenplay/src/screenplay/service.rs
use rig::{agent::Agent, completion::Prompt, providers::{self, openai::CompletionModel}};
use schemars::schema_for;

use super::{models::Screenplay, utils::create_response_format};

static PREAMBLE: &str = r#"
You are a screenplay writer for YouTube videos. You will be given a prompt, and you will write a screenplay based on the prompt.
Make it engaging and non-trivial. Try to find interesting facts and characters to write about.
"#;
pub struct ScreenplayService {
    agent: Agent<CompletionModel>,
}

impl ScreenplayService {
    pub fn new() -> Self {

        let client = providers::openai::Client::from_env();

        let schema = schema_for!(Screenplay);

        let response_format = create_response_format(&schema, "screenplay_schema");

        let agent = client
            .agent("gpt-4o")
            .preamble(PREAMBLE)
            .additional_params(response_format)
            .build();

        Self { agent }
    }

    pub async fn generate_screenplay(&self, prompt: &str) -> String {

        let response = self.agent
        .prompt(prompt)
        .await
        .unwrap();

        response
    }
}

I send another request and see the following output:

Schema tweaks

The thing is that OpenAI APIs have some restrictions regarding the schema you can use in the request. From the error above, we see that addintional_properties needs to be present in the schema and set to false. All restrictions are described in the documentation

From my perspective, the biggest restriction is that the OneOff property is not allowed. In Rust, it is common to use enums when defining domain, but it won't work with OpenAI API.

I would like to return a response similar to this

#[derive(Serialize, Deserialize)]
pub struct ScreenplayResponse {
    pub title: String,
    pub scenes: Vec<Scene>,
}

#[derive(Serialize, Deserialize)]
pub enum Scene {
    StudioScene(StudioScene),
    CutToScene(CutToScene),
    FadeOutScene(FadeOutScene),
}

// ....

I can't create JSON schema directly from this struct. My workaround is to have a different object as a response_format for the model. I will map it later to my expected shape.
To avoid OneOff limitations, I created a struct without enums

I use the newest alpha version of schemars 1.x library because I wanted to play with the functionalities it offers.

#[derive(Serialize, Deserialize, JsonSchema)]
#[schemars(transform = add_additional_properties)]
pub struct Screenplay {
    pub title: String,
    pub studio_scenes: Vec<StudioScene>,
    pub cut_to_scenes: Vec<CutToScene>,
    pub fade_out_scenes: Vec<FadeOutScene>,
}

#[derive(Serialize, Deserialize, JsonSchema)]
#[schemars(transform = add_additional_properties)]
pub struct StudioScene {
    #[schemars(transform = sanitize_number)]
    pub scene_number: i32,
    pub location: String,
    pub time: String,
    pub description: String,
    pub dialogue: Vec<Dialogue>,
}

// ...

My hope is that I can receive the output anyway and sort it later on. Let's try

Ok, quite decent result!

Deployment

Let's run cargo lambda build -r

ooops.

Ok, it is a very common issue. The solution is not to use openssl, or to add openssl as a package with a vendored flag.

Sometimes, the second option is the only one if a library you are using doesn't let you differ from nativetls. Luckily, the current version of rig allows us to use reqwest-rustls flag. After checking the default flags in dosc, I see that I also need to disable default features to get rid of the default reqwest version. Here is the final Cargo.toml for my screenplay library

[package]
name = "screenplay"
version = "0.1.0"
edition = "2021"

[dependencies]
rig-core = { version = "0.10.0", features = ["reqwest-rustls"], default-features = false }
schemars = { version = "1.0.0-alpha.17", features = ["schemars_derive"] }
serde = { version = "1.0.219", features = ["derive"] }
serde_json = "1.0.140"

Now build should work seamlessly

For small POC and prototyping, cargo lambda deployment works great. I define timeout and env properties in the handler's Cargo.toml

[package]
name = "screenplay_handler"
version = "0.1.0"
edition = "2021"

[dependencies]
lambda_http = "0.13.0"

tokio = { version = "1", features = ["macros"] }
screenplay = { path = "../screenplay" }
serde = "1.0.219"
serde_json = "1.0.140"

[package.metadata.lambda.deploy]
timeout = 120
env_file = ".env"

and run

cargo lambda deploy --enable-function-url

Now I have a URL to test the function

Testing

I use Bruno to test my function

Looks good!

Summary

In this blog, I created a simple Lambda Function that calls OpenAI API and uses response_format to make sure that the result has an expected shape. The rich Rust's ecosystem, including serde and schemars makes it easy to create a typesafe bridge between LLM's answer and our system.

Having the typesafe output from LLM simplifies the pipeline and allows seamless utilization. It also allows estimating tokens' usage more precisely, as no additional steps for formatting text are needed.

Async events with AppSync Events API

szymon-szym — Sun, 09 Feb 2025 15:54:37 +0000

Photo by Dawn McDonald on Unsplash

Simple setup for live events with AppSync Events

Scenario

Let's build a simple system, where Users can log in and request a report generation. Generating a report takes a while, and we don't want them to wait for it. We will let them know once the report is generated.

What we are building

To achieve this goal I will utilize AppSync Evetns API which is a fairly new service that allows implement WebSockets in an easy way.

Architecture

AppSync Events API contains two main pieces - channels for asynchronous communication, and HTTP endpoint to publish events to channels.

The user will use the web page to log into the system, and connect to the channel generated-repots/<user-id>. It is important that the user can see only its own reports.

To generate a new report user calls generate-report endpoint on the API Gateway. From there, the lambda function extracts user ID from the authorizer context and triggers StepFunction, which simulates the long-running task. Once the job is done, StepFunction publishes the information about it to the generated-repots/<user-id> channel.

Implementation

The code is available IN THIS REPO

Authentication

For auth I am using Cognito.

With AWS CDK, setting up the Cognito is quite straightforward. The only tricky part is that we need to define a default authenticated role and a default policy.

// lib/appsync_events-stack.ts

// ...
    const cognitoUserPool = new cdk.aws_cognito.UserPool(this, "EventsApp", {
      selfSignUpEnabled: true,
      signInAliases: { email: true },
      autoVerify: { email: true },
      passwordPolicy: {
        minLength: 8,
        requireLowercase: true,
        requireUppercase: true,
        requireDigits: true,
      },
    });

    const cognitoUserPoolClient = new cdk.aws_cognito.UserPoolClient(
      this,
      "EventsAppClient",
      {
        userPool: cognitoUserPool,
        generateSecret: false,
        authFlows: {
          adminUserPassword: true,
          userSrp: true,
        },
      }
    );

    const cognitoIdentity = new cdk.aws_cognito.CfnIdentityPool(
      this,
      "EventsAppIdentityProvider",
      {
        allowUnauthenticatedIdentities: false,
        cognitoIdentityProviders: [
          {
            clientId: cognitoUserPoolClient.userPoolClientId,
            providerName: cognitoUserPool.userPoolProviderName,
          },
        ],
      }
    );


    const authenticatedRole = new cdk.aws_iam.Role(
      this,
      "CognitoDefaultAuthenticatedRole",
      {
        assumedBy: new cdk.aws_iam.FederatedPrincipal(
          "cognito-identity.amazonaws.com",
          {
            StringEquals: {
              "cognito-identity.amazonaws.com:aud": cognitoIdentity.ref,
            },
            "ForAnyValue:StringLike": {
              "cognito-identity.amazonaws.com:amr": "authenticated",
            },
          },
          "sts:AssumeRoleWithWebIdentity"
        ),
      }
    );

    const defaultPolicy = new cdk.aws_cognito.CfnIdentityPoolRoleAttachment(
      this,
      "IdentityPoolRoleAttachment",
      {
        identityPoolId: cognitoIdentity.ref,
        roles: {
          authenticated: authenticatedRole.roleArn,
        },
      }
    );
//...

AppSync Events

For the moment I am writing this post, there are no constructs for AppSync Events for AWS CDK yet. This is not a blocker, as there is a CloudFormation support, so I can use L1 constructs inside CDK. In other words, I can write CloudFormation inside my typescript CDK stack definition.

The initial declaration contains only some information about auth providers:

// lib/appsync_events-stack.ts

//...
    const eventsAPI = new cdk.aws_appsync.CfnApi(this, "MyEventsAPI", {
      name: "MyEventsAPI",
      eventConfig: {
        authProviders: [
          {
            authType: "API_KEY",
          },
          {
            authType: "AMAZON_COGNITO_USER_POOLS",
            cognitoConfig: {
              userPoolId: cognitoUserPool.userPoolId,
              awsRegion: "us-east-1",
            },
          },
        ],
        connectionAuthModes: [
          {
            authType: "AMAZON_COGNITO_USER_POOLS",
          },
        ],
        defaultPublishAuthModes: [
          {
            authType: "API_KEY",
          },
          {
            authType: "AMAZON_COGNITO_USER_POOLS",
          },
        ],
        defaultSubscribeAuthModes: [
          {
            authType: "AMAZON_COGNITO_USER_POOLS",
          },
        ],
      },
    });
// ...

I picked two flows. Cognito user pool is for users, and API key is for my StepFunction.

It is almost it. I need to define the channel namespace and API key

// lib/appsync_events-stack.ts

//...
    const generatedRequestNamespace = new cdk.aws_appsync.CfnChannelNamespace(
      this,
      "GeneratedRequestNamespace",
      {
        apiId: eventsAPI.attrApiId,
        name: "generated-reports",
      }
    );

    const eventsApiKey = new cdk.aws_appsync.CfnApiKey(this, "EventsApiKey", {
      apiId: eventsAPI.attrApiId,
    });
// ...

StepFunction

My state machine is fairly simple and contains two steps: waiting, and publishing the event.

To do the latter I will utilize Call HTTPs APIs action, which is the absolute killer feature for StepFunctions, as it allows integration with any API-based service within or outside AWS.

To be able to call the endpoint securely, I need to define a Connection (the same we would use for the API Destinations in the EventBridge)

// lib/appsync_events-stack.ts

// ...
    const eventsAPIKeySecret = new cdk.aws_secretsmanager.Secret(
      this,
      "EventsApiKeySecret"
    );

    const eventsConnection = new cdk.aws_events.Connection(
      this,
      "EventsAPIConnection",
      {
        authorization: cdk.aws_events.Authorization.apiKey(
          "x-api-key",
          cdk.SecretValue.secretsManager(eventsAPIKeySecret.secretName)
        ),
      }
    );
// ...

Here is where my code gets a bit clunky - I need to have Secret defined to be used for Connection, even though under the hood, the Connection creates its own secret in the SecretManager. I haven't figured out yet how to get rid of the spare custom Secret.

Here are the tasks defined for the state machine. For anything larger than a small demo I would use a separate .asl.json file to store the StepFunction definition. I almost always debug and update the StepFunction in the AWS console, and download the updated definition to keep it in the source control. This approach has its pros and cons, but this is a topic for another discussion.

In this example, tasks are defined in the CDK

// lib/appsync_events-stack.ts

//...
const waitTask = new cdk.aws_stepfunctions.Wait(this, "ComplexTask", {
      time: cdk.aws_stepfunctions.WaitTime.duration(cdk.Duration.seconds(30)),
    });

    const publishToAppsync = new cdk.aws_stepfunctions_tasks.HttpInvoke(
      this,
      "PublishToAppSync",
      {
        apiEndpoint: cdk.aws_stepfunctions.TaskInput.fromText("/event"),
        method: cdk.aws_stepfunctions.TaskInput.fromText("POST"),
        connection: eventsConnection,
        body: cdk.aws_stepfunctions.TaskInput.fromJsonPathAt("$.input"),
        apiRoot: "https://" + eventsAPI.getAtt("Dns.Http").toString(),
      }
    );
// ...

The StepFunction definition brings it all together

// lib/appsync_events-stack.ts

// ...
    const longTaskSfn = new cdk.aws_stepfunctions.StateMachine(
      this,
      "LongTaskStateMachine",
      {
        definition: waitTask.next(publishToAppsync),
        stateMachineType: cdk.aws_stepfunctions.StateMachineType.EXPRESS,
        logs: {
          destination: new cdk.aws_logs.LogGroup(
            this,
            "LongTaskStateMachineLogGroup"
          ),
          level: cdk.aws_stepfunctions.LogLevel.ALL,
          includeExecutionData: true,
        },
      }
    );
// ...

API Gateway

API HTTP Gateway will handle a single endpoint and use Cognito authorizer to authorize requests.

// lib/appsync_events-stack.ts

// ...
const httpApi = new cdk.aws_apigatewayv2.HttpApi(this, "HttpApi", {
      corsPreflight: {
        allowOrigins: ["*"],
        allowMethods: [cdk.aws_apigatewayv2.CorsHttpMethod.ANY],
        allowHeaders: ["*"],
      },
    });

    const cognitoAuthorizer =
      new cdk.aws_apigatewayv2_authorizers.HttpUserPoolAuthorizer(
        "CognitoAuthorizer",
        cognitoUserPool,
        {
          userPoolClients: [cognitoUserPoolClient],
        }
      );
// ...

I need a Lambda function to handle the request. Lambda needs permissions to invoke StepFunction

// lib/appsync_events-stack.ts

// ...
    const reportHandlerFunction = new rustLambda.RustFunction(this, "reportHandlerFunction", {
      entry: ("http_handler"),
      environment: {
        STEP_FUNCTION_ARN: longTaskSfn.stateMachineArn,
      },
    });

    longTaskSfn.grantStartExecution(reportHandlerFunction);

    httpApi.addRoutes({
      path: "/generate-report",
      methods: [HttpMethod.POST],
      integration:
        new cdk.aws_apigatewayv2_integrations.HttpLambdaIntegration(
          "reportHandlerIntegration",
          reportHandlerFunction
        ),
      authorizer: cognitoAuthorizer,
    });
// ...

Lambda Function - code

Once the infrastructure is defined, let's create a Lambda function to handle the requests.

// http_handler/src/main.rs
use aws_config::BehaviorVersion;
use lambda_http::{run, service_fn, tracing, Body, Error, Request, RequestExt, Response};
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize)]
struct Message {
    message: String,
}

#[derive(Serialize, Deserialize)]
struct AppSyncEventsRequest {
    channel: String,
    events: Vec<String>,
}

#[derive(Serialize, Deserialize)]
struct StepFunctionInput {
    input: AppSyncEventsRequest,
}

async fn function_handler(
    sfn_client: &aws_sdk_sfn::Client,
    event: Request,
) -> Result<Response<Body>, Error> {

    let machine_arn = std::env::var("STEP_FUNCTION_ARN").unwrap();

    let context = event.request_context();

    let authorizer_data = context.authorizer().unwrap();

    let jwt_data = authorizer_data.jwt.clone().unwrap();

    let user_name = jwt_data.claims.get("username").unwrap();

    let message = Message {
        message: "report generated".to_string(),
    };

    let step_function_input = StepFunctionInput {
        input: AppSyncEventsRequest{
            channel: format!("generated-reports/{user_name}"),
            events: vec![serde_json::to_string(&message).unwrap()],
        }
    };

    let result = sfn_client
        .start_execution()
        .set_input(Some(serde_json::to_string(&step_function_input).unwrap()))
        .state_machine_arn(machine_arn)
        .send()
        .await
        .unwrap();

    let message = format!("{:?}", result);

    let resp = Response::builder()
        .status(200)
        .header("content-type", "text/html")
        .body(message.into())
        .map_err(Box::new)?;
    Ok(resp)
}

#[tokio::main]
async fn main() -> Result<(), Error> {
    tracing::init_default_subscriber();

    let config = aws_config::load_defaults(BehaviorVersion::latest()).await;

    let sfn_client = aws_sdk_sfn::Client::new(&config);

    run(service_fn(|ev| function_handler(&sfn_client, ev))).await
}

UI

I've created a new Vue3 application. I will use Amplify library which makes integration with Cognito and AppSync very easy. It can be used even if you are not planning to use Amplify service to deploy your frontend.

To configure Cognito and AppSync Events I add in the main.ts

// fe/src/main.ts
// ...
import { Amplify } from "aws-amplify";

Amplify.configure({
  API: {
    Events: {
      endpoint:
        "https://<your_endpoint>.appsync-api.us-east-1.amazonaws.com/event",
      region: "us-east-1",
      defaultAuthMode: "userPool",
    },
  },
  Auth: {
    Cognito: {
      userPoolId: "<user_pool>",
      userPoolClientId: "<client_id>",
      identityPoolId: "<identity_id>",
      loginWith: {
        email: true,
      },
      signUpVerificationMethod: "code",
      userAttributes: {
        email: {
          required: true,
        },
      },
      allowGuestAccess: true,
      passwordFormat: {
        minLength: 8,
        requireLowercase: true,
        requireUppercase: true,
        requireNumbers: true,
        requireSpecialCharacters: true,
      },
    },
  },
});
// ...

I wrap the whole application in the authenticator tag in the App.vue

// fe/src/App.vue
// ...
import { Authenticator } from '@aws-amplify/ui-vue'
import "@aws-amplify/ui-vue/styles.css";
</script>

<template>
  <authenticator>
    <div class="wrapper">
      <HelloWorld msg="You did it!" />
    </div>
  </authenticator>
</template>
// ...

To connect to the channel, it is enough to call events.connect

// fe/src/components/HelloWorld.vue
// ...
import { events } from 'aws-amplify/data'
const { user } = useAuthenticator();

onMounted(async () => {

  const channel = await events.connect(`/generated-reports/${user.username}`);

  channel.subscribe({
    next: (data) => {
      console.log('received', data.event);
      reports.value.push(data.event)
    },
    error: (err) => console.error('error', err),
  });

})
// ...

In the real world, you would move the connection part to a separate composable or keep it in the store, but for now, it is enough to keep it directly in the component.

Deploy

Once CDK is deployed, we need to make sure that our Connection has the right API key. Grab the key from the AppSync Events console

and update it in the Connections section in the EventBridge

Test

Start the frontend

cd fe
npm run dev

There is a default Amplify login form. Sign up for the service and log in.

After login, I can see my amazing UI, which contains from the user-id and a single button to generate report

I can check in the dev tools and see, that there is a WebSockets connection opened.

After clicking the button I need to wait for ~30 seconds, and I receive a message from the channel I subscribed for

It works 🎉 🎉

Summary

AppSync Events API allows the setup of a WebSocket connection with minimal configuration overwhelm. It lets publishing events via HTTPs endpoint, which any authorized service, including StepFunction can use.

Amplify library simplifies setting up the authorized connection from the UI to Events API.

Managing users’ permissions with Cedar policies in the AWS Lambda

szymon-szym — Mon, 06 Jan 2025 10:48:57 +0000

Photo by Dmitrii Pashutskii on Unsplash

You can define complex permission rules with human-readable policies and manage them without changing the application code.

Introduction

The blog post shows how to define and use Cedar policies in the Lambda Function. We will define policies using Cedar and use cesar-policy crate to validate them in the Lambda Function code.

Scenario

We have a service for managing generated reports. Reports are stored as PDF files on the S3. Users can see general info about reports and download files using presigned URLs. While general data related to reports is available for all authenticated users, the presigned URLs can be generated only by the report owners.

What we are building

I create a REST API that exposes two endpoints: get-report-info and get-report-url. The first is available for all authenticated users, the second is only for reports' owners.

Architecture

The high-level picture of the overall solution could look like this.
The green rectangle highlights the part that will be implemented in this blog post.

Tech stack

IaC - AWS SAM
Application code - Rust, cargo-lambda
Authorization logic - Cedar, cedar-policy

The whole code is available in this repository

Cedar

Cedar is the language for defining authorization policies and making decisions based on them. The main idea is to decouple authorization logic from the business logic of the application.

The syntax of Cedar is simple yet powerful. It allows for defining fine-grained access restrictions in a human-readable fashion. Let's see our example:

permit(
    principal is User,
    action == Action::"GetReportInfo",
    resource == Resource::"ReportData"
);
permit(
    principal is User,
    action == Action::"SetReportInfo",
    resource == Resource::"ReportData"
) when {
    resource.owner_id == principal.id
};
permit(
    principal is User,
    action == Action::"GenerateS3Url",
    resource == Resource::"S3Object"
) when {
    resource.owner_id == principal.id
};

Even if this is the first time you see Cedar policy, you probably can reason about it.

There are 3 policies, for 3 different actions:

GetReportInfo is available for every User (principal) for all ReportData (resource)
SetReportInfo allows only users who are owners of the resource
the same goes for GenerateS3Url

Authorization check

The policies alone don't do much. To be able to run authorization verification I need a few more elements.

Entities

The entities are principal, resource, and action. Authorization logic uses them to validate requests against policies set.

Context

Sometimes you would need to attach additional information that is not related to the specific user, like source IP, or time of request. The context is the right place to include this data.

Schema

I won't use schema validation, however, it is a powerful tool to guarantee policies' correctness. More about this topic can be found in documentation

Implementation

The authorization logic is decoupled from the application code, which means that we need to map the application logic to Cedar entities.

For my simple use case, I create a service in Rust to handle authorization and use it directly in my Lambda code. I could also build a Lambda Extension and use it from a Lambda Function written in any language.

Models

To make a decision, my authorization service would need to receive the user id, action type, resource type and resource owner id. I use a rich Rust's types system to describe expected input.

// authorization/models.rs


// input types

#[derive(Clone, Debug)]
pub struct UserId(pub String);

impl UserId {
    pub fn as_str(&self) -> &str {
        &self.0
    }
}

#[derive(Clone, Debug)]
pub struct OwnerId(pub String);

impl OwnerId {
    pub fn as_str(&self) -> &str {
        &self.0
    }
}


#[derive(Serialize, Deserialize, Clone)]
pub enum ActionVerb {
    GetReportInfo,
    SetReportInfo,
    GenerateS3Url
}


impl fmt::Display for ActionVerb {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        let str = match self {
            ActionVerb::GetReportInfo => "GetReportInfo",
            ActionVerb::SetReportInfo => "SetReportInfo",
            ActionVerb::GenerateS3Url => "GenerateS3Url"
        };
        write!(f, "{}", str)
    }
}

#[derive(Serialize, Deserialize, Clone)]
pub enum ResourceType {
    S3Object,
    ReportData
}

impl fmt::Display for ResourceType {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        let str = match self {
            ResourceType::S3Object => "S3Object",
            ResourceType::ReportData => "ReportData"
        };

        write!(f, "{}", str)
    }
}

// output

#[derive(Debug)]
pub enum AuthorizationDecision {
    Allow,
    Deny
}

UserId and OwnerId utilize a newtype pattern. I also implement as_str for both of them to make it easy to get a reference to their inner values.

ActionVerb and ResourceType are enums with implemented Display to easily convert to_string and print to the console if needed.

There are also internal types to help with creating cedar-policy entities.

// models.rs
// ...

// internal types

#[derive(Clone)]
pub struct UserEntityInput {
    pub user_id: UserId
}

impl UserEntityInput {
    pub fn new(user_id: UserId) -> Self {
        Self { user_id }
    }
}

#[derive(Clone)]
pub struct ActionEntityInput {
    pub verb: ActionVerb
}

impl ActionEntityInput {
    pub fn new(verb: ActionVerb) -> Self {
        Self { verb }
    }
}

#[derive(Clone)]
pub struct ResourceEntityInput {
    pub resource_type: ResourceType,
    pub owner_id: OwnerId
}

impl ResourceEntityInput {
    pub fn new(resource_type: ResourceType, owner_id: OwnerId) -> Self {
        Self { resource_type, owner_id }
    }
}

#[derive(Clone)]
pub enum CreateEntityInput {
    User(UserEntityInput),
    Action(ActionEntityInput),
    Resource(ResourceEntityInput)
}

Authorize function

The main function of the authorization service takes input defined above, and returns decision.

// service.rs

//...
pub(crate) fn authorize(
    user_id: UserId,
    action_verb: ActionVerb,
    resource_type: ResourceType,
    owner_id: OwnerId,
) -> Result<AuthorizationDecision, Box<dyn Error>> {

    let policy = get_policy_set()?;

    let user_entity_input = UserEntityInput::new(user_id);

    let action_entity_input = ActionEntityInput::new(action_verb);

    let resource_entity_input = ResourceEntityInput::new(resource_type, owner_id);

    let principal = create_entity(CreateEntityInput::User(user_entity_input))?;

    let action = create_entity(CreateEntityInput::Action(action_entity_input))?;

    let resource = create_entity(CreateEntityInput::Resource(resource_entity_input))?;

    let request = Request::new(
        principal.uid(),
        action.uid(),
        resource.uid(),
        Context::empty(),
        None,
    )?;

    let entities = Entities::from_entities([principal, action, resource], None)?;

    let authorizer = Authorizer::new();

    let answer = authorizer.is_authorized(&request, &policy, &entities);

    let decision = match &answer.decision() {
        cedar_policy::Decision::Allow => {
            println!("Allowed: {:?}", &answer);
            AuthorizationDecision::Allow
        },
        cedar_policy::Decision::Deny => {
            println!("Deny: {:?}", &answer);
            AuthorizationDecision::Deny
        },
    };

    Ok(decision)
}
// ...

Let's go through it step by step:

fetch policies with get_policy_set function. In this example, policies are defined as a string directly in the code. In real life, policies can be stored in the S3, DynamoDB, or wherever.
convert received function args to the internal type CreateEntityInput
create cedar-policy entities using my create_entity helper function
prepare Request to the Authorizer, and check the decision for the given input

Using inside Lambda Function

The flow of authorizing requests will be the following:

At this point, I will mock my reports service and return reports from the static list. I would use DynamoDB or any other database here in the real scenario.

I've also made some assumptions regarding the Claims shape in the JWT token. They would work if you use Cognito but might not work with other services.

Get S3 URL hanlder

The authorization flow in the handler looks like this:

// handlers/get_report_s3url.rs

// ...

pub(crate) async fn function_handler(
    event: Request,
    reports_service: &ReportsService,
) -> Result<Response<Body>, Error> {

    // get user id from auth token
    let token = event
        .headers()
        .get(AUTHORIZATION)
        .and_then(|header| header.to_str().ok())
        .and_then(|token| token.strip_prefix("Bearer "))
        .ok_or("Missing auth token")?;

    let user_id = user_context::jwt::get_user_id(token)?;

    // get report id

    let report_id = event
        .query_string_parameters_ref()
        .and_then(|params| params.first("report"))
        .ok_or("Missing report id")?;


    // get report

    let report = reports_service.get_report(report_id.to_string())
        .ok_or("missing report with given id")?;

    // authorize request
    let decision = authorization::service::authorize(
        UserId(user_id),
        ActionVerb::GenerateS3Url,
        ResourceType::S3Object,
        OwnerId(report.owner_id),
    )?;


    // return response

    let resp = match decision {
        AuthorizationDecision::Allow => Response::builder()
            .status(200)
            .header("content-type", "text/html")
            .body("Here you go, this is your dummy s3 url: dummy".into())
            .map_err(Box::new)?,
        AuthorizationDecision::Deny => Response::builder()
            .status(404)
            .header("content-type", "text/html")
            .body("You are not authorized".into())
            .map_err(Box::new)?,
    };
    Ok(resp)
}

extract a token from the authorization header
get the user id from the jwt token using get_user_id function. There is an assumption that claims include username field which is the user id. The important thing is that I didn't validate the token, because it was already validated on the API HTTP Gateway.
get the report id from the query string and look for the report using reports_service
use user id and owner from the report to get authorization decision

Test handler

In the root folder, I run the following command:

sam build && sam local start-api

Once the local container is up, I test the path using Bruno. (I use the token from the already configured cognito user pool from another project, with username in Claims 04781408-1081-706c-c3ac-3c618d5a379a). My dummy data in the ProjectsService looks like this:

//...
reports: vec![
                Report {
                    id: "abc-123".to_string(),
                    title: "dummy report".to_string(),
                    owner_id: "04781408-1081-706c-c3ac-3c618d5a379a".to_string(),
                    s3_key: "dummy/key/file.pdf".to_string(),
                },
                Report {
                    id: "abc-124".to_string(),
                    title: "dummy report 2".to_string(),
                    owner_id: "123".to_string(),
                    s3_key: "dummy/key/file.pdf".to_string(),
                },
                Report {
                    id: "abc-125".to_string(),
                    title: "dummy report 3".to_string(),
                    owner_id: "321".to_string(),
                    s3_key: "dummy/key/file.pdf".to_string(),
                },
            ]
// ...

I expect to be able to get URL for the abc-123 report and be declined for others.

Looks good!

In the get_report_data handler the flow is exactly the same, there is only different action and resource type defined

// handlers/get_report_data.rs

// ...
let decision = authorization::service::authorize(
        UserId(user_id),
        ActionVerb::GetReportInfo,
        ResourceType::ReportData,
        OwnerId(report.owner_id),
    )?;
// ...

As expected this time I am able to query data for abc-124 report

It works 🎉 🎉 🎉

Summary

In this blog post, I created the authorization flow using the Cedar language to define policies and the cedar-policy crate to run authorization logic in the Rust code.

Here are my thoughts after playing with the Cedar language and using it in the AWS Lambda with cedar-policy crate

Cedar is a powerful and expressive language. It allows you to define access control based on role, attributes, and relations.
Cedar was created with complex systems in mind. It allows managing thousands of policies and complex access patterns, which require solving different problems, such as managing and updating policies.
Simpler solutions can still benefit from decoupling authorization logic from application code. Policies can be stored in S3 or any db.
Authorization logic can be seamlessly integrated into Rust code using cedar-policy crate.

Forem: szymon-szym

Run my own serverless service on a Raspberry Pi with Spin

Idea

Architecture

Goal

Application development

Add SQLite

Deploy to Raspberry Pi

Summary

Safe AWS Lambda Durable Functions with XState

Background

Problem

Goal

Experiment

Implementation

State Machine

Actors

State Machine

Type safety

Visual representation

Run on Durable Functions

Summary

Streaming AI Agent response with AWS REST Gateway and Lambda Function

Goal

Architecture

Lambda Function

Infrastructure

Testing

Summary

Deploy Rust Agent to AWS AgentCore Runtime with GitHub actions

Goal

Architecture

Agent

/Invocations

Main function

Deployment

Dockerfile

Infratructure

Deployment pipeline

Set up AWS credentials and install CDK

Deploy ECR repository

Build the image

Deploy AgentCore Runtime

Bump agent version

Finding out

MicroVM Metadata Service (MMDS)

Resources provisioning

Testing

Next steps

Summary

Agent conversation context with AWS Bedrock AgentCore Memory

Goal

Implementation

Cognito

Agent

Prepare AgentCore config

Provide cloud resources

Testing

Getting Cognito token

Local test with Bruno

Testing live endpoint

Summary

Serverless Semantic Search: A Hands-On Guide to AWS S3 Vector Buckets with Rust

Introduction

Goal

Architecture

Creating a bucket

Populate data

Query data with Lambda

Deploy and test

Summary

MCP OAuth on AWS Lambda with WorkOS

Goal

Problem

Architecture

Implementation

Initial setup

Local testing

Authorization Server

/.well-known/oauth-authorization-server

Add `SQLite`

`/.well-known/oauth-authorization-server`

`/register`

`/authorize`

`/callback`

`/token`