DEV Community: ohalay

Integrating MCP Tools with AWS Bedrock in an ASP.NET Core Minimal API

ohalay — Wed, 05 Nov 2025 19:09:04 +0000

In this article, we'll build an ASP.NET Core Minimal API that integrates AWS Bedrock with MCP client capabilities, enabling the dynamic invocation of AI tools through standardized MCP interfaces.

How to work with LLM models

There are several ways to interact with LLM models in .NET:

Direct LLM provider client (API or SDK) - lower-level abstraction
Microsoft.Extensions.AI - middle-level abstraction
Semantic Kernel - higher-level abstraction

We'll use the mid-level abstraction (Microsoft.Extensions.AI), which balances simplicity and flexibility while keeping the implementation decoupled from specific model providers. To start working, we need to install dependencies:

AI extension itself
Bedrock AI

Then register IChatClient abstraction

services.AddDefaultAWSOptions(builder.Configuration.GetAWSOptions());
services.TryAddAWSService<IAmazonBedrockRuntime>();
services.AddSingleton<IChatClient>(sp =>
{
  var runtime = sp.GetRequiredService<IAmazonBedrockRuntime>();
  var chatClient = runtime.AsIChatClient().AsBuilder().UseFunctionInvocation().Build();
  return chatClient;
}

Add MCP client

The IChatClient interface supports AI tools that extend its capabilities. We can dynamically load these tools from an MCP server.

var httpClient = new HttpClient();
var transport = new HttpClientTransport(new
{
    Endpoint = new Uri("https://my-mcp-server-enpoint.com"),
    Name = "Mcp Client",
}, httpClient);

var mcpClient = await McpClient.CreateAsync(transport);
var mcpTools = await mcpClient.ListToolsAsync();

A possible MCP server configuration looks like this:

ohalay

Sep 18 '25

Add the MCP server to the ASP.NET Core minimal API

#mcp #tutorial #dotnet #ai

Comments 2

2 min read

Chat API

Now, let’s combine everything into a single endpoint /api/message that sends user messages to the LLM while using MCP tools.

app.MapPost("/api/message", async (ChatRequest request, IChatClient chatClient) =>
{
  ...
  var response = await chatClient.GetResponseAsync(
    [new ChatMessage(ChatRole.User, request.Message)],
    new ChatOptions
    {
        ModelId = "anthropic.claude-3-5-sonnet-20240620-v1:0",
        Tools = [.. mcpTools],
    }
);

  return Results.Ok(response);
})

In the end, to authenticate the user request to MCP, we will use an HTTP handler for our transport

public class OnBehalfOfHttpHandler(IHttpContextAccessor httpContextAccessor) : DelegatingHandler
{
  override protected Task<HttpResponseMessage> SendAsync(HttpRequestMessage request, CancellationToken cancellationToken)
  {
    var authorizationHeader = httpContextAccessor.HttpContext?.Request.Headers.Authorization.FirstOrDefault();
    if (authorizationHeader is not null && AuthenticationHeaderValue.TryParse(authorizationHeader, out var headerValue))
    {
      request.Headers.Authorization = new AuthenticationHeaderValue("Bearer", headerValue.Parameter);
    }
    return base.SendAsync(request, cancellationToken);
  }
}

Solution Limitation

Only Bedrock models supported by Microsoft.Extensions.AI can be used.
Some models may not support tool invocation
The MCP SDK is still in preview and subject to change

Conclusion

By combining IChatClient with MCP tools, you create a modular and extensible architecture for AI integration in .NET.
This approach lets you:

Keep your code provider-agnostic
Dynamically load new AI tools via MCP
Securely delegate authentication using "on-behalf-of" tokens

As MCP and Microsoft.Extensions.AI evolve, this pattern positions your application to easily adopt new models, providers, and AI capabilities with minimal refactoring.

How to Build Semantic Search in ASP.NET Core using PostgreSQL

ohalay — Mon, 06 Oct 2025 19:09:33 +0000

I usually build simple applications that expose public APIs backed by PostgreSQL databases. But sometimes, it is not enough for the product. However, businesses often need more flexible and intelligent search capabilities than what PostgreSQL’s built-in search provides. Someone may argue that PostgreSQL supports full-text search. Yes, it's fine, but it's more for finding specific words or phrases. Semantic search focuses on understanding contextual meaning rather than exact keyword matches - and that’s what we’ll explore.

Database side

The first solution that comes to mind is Elasticsearch. It’s an excellent tool for our problem, but it introduces extra cost, additional infrastructure, and the need to learn and integrate a new technology. Since we already use PostgreSQL, we can simply enable the pgvector extension for vector search.

Install NuGet to support vectors
Configure a model to store and search vectors. I'm using the ivfflat index type, which is optimised for approximate nearest neighbor search. And I'm enabling vector_cosine_ops cosine distance similarity search.

modelBuilder.HasPostgresExtension("vector");

modelBuilder.Entity<TextEmbedding>(entity =>
{
  entity.HasKey(e => e.Id);
  entity.Property(e => e.Content).IsRequired();
  // all-minilm produces 384-dimensional vectors
  entity.Property(e => e.Embedding).HasColumnType("vector(384)"); 
  entity.Property(e => e.CreatedAt).IsRequired();

  // Add index for vector similarity search
  entity.HasIndex(e => e.Embedding)
   .HasMethod("ivfflat")
   .HasOperators("vector_cosine_ops");
});

Application part

Now that the database is ready to work with vectors, let’s move to the application side. We'll register an embedding service, create an endpoint to store text embeddings, and finally build a semantic search endpoint.

1. Register the embedding service.

Microsoft provides an abstraction layer for working with large language models via the Microsoft.Extensions.AI NuGet package, so we don’t have to depend on a specific provider. I will use Ollama - a free, open-source tool that simplifies running LLMs. NuGet OllamaSharp to work in NET.

services.AddSingleton<IEmbeddingGenerator<string, Embedding<float>>>(sp =>
{
  var modelId = builder.Configuration["Ollama:EmbeddingModel"];
  var baseUrl = builder.Configuration["Ollama:BaseUrl"];
  return new OllamaApiClient(baseUrl, modelId);
});

2. Create an endpoint to convert text to vectors and store them

app.MapPost("/text", async (
  Text[] request, 
  EmbeddingDbContext context, 
  IEmbeddingGenerator<string, Embedding<float>> embeddingService) =>
{
  // Convert request to vectors
  var embeddings = await embeddingService.GenerateAsync(
    request.Select(s => s.Content)
  );

  var dbModels = embeddings.Select((embedding, index) => new TextEmbedding
  {
    Content = request[index].Content,
    Embedding = new Vector(embedding.Vector),
    CreatedAt = DateTime.UtcNow
  }).ToList();

  context.AddRange(dbModels);
  await context.SaveChangesAsync();

  return TypedResults.Created();
});

3. The magic happens here - semantic search using cosine similarity

app.MapGet("/text/search", async (
  string query, 
  EmbeddingDbContext context, 
  IEmbeddingGenerator<string, Embedding<float>> embeddingService) =>
{
  int limit = 5;

  // Convert query to vector
  var embeddings = await embeddingService.GenerateAsync(query);
  var queryEmbedding = new Vector(embeddings.Vector);

  // Find most similar texts using cosine distance
  var similarTexts = await context.TextEmbeddings
    .OrderBy(x => x.Embedding.CosineDistance(queryEmbedding))
    .Select(x => new Text(x.Content))
    .Take(limit)
    .ToListAsync();

  return TypedResults.Ok(similarTexts);
});

Test local

To run everything locally, we can use Ollama for generating embeddings. This approach avoids cloud APIs (like OpenAI or Azure) and keeps everything self-contained.

ollama:
  image: ollama/ollama:latest
  container_name: ollama
  ports:
    - "11434:11434"
  entrypoint:
    - "/bin/bash"
    - "-c"
    - "ollama serve & sleep 5 && ollama pull all-minilm && wait"
  environment:
    - OLLAMA_KEEP_ALIVE="24h"
  volumes:
    - ollama_data:/root/.ollama

The rest of the stack - the API, PostgreSQL, and database migrations - can be managed within the same Docker Compose setup.

Conclusion

By combining EF Core, PostgreSQL, and the pgvector extension, we can bring powerful semantic search capabilities directly into ASP.NET applications — without introducing new infrastructure.
The approach is efficient, cost-effective, and fully compatible with familiar .NET tools. Thanks to the Microsoft.Extensions.AI abstraction, we can easily swap between local (Ollama) and cloud (OpenAI, Azure) embeddings. This setup is ideal for building document search, recommendation engines, similarity detection, and RAG-based applications.
You can find the full implementation details on my

ohalay / embedeting-poc

Architecture

flowchart LR
    subgraph Architecture
        API[API .NET]
        DB[(PostgreSQL using pgvector)]
        OLLAMA[Ollama LLM]
        MIGRATIONS[EF Migrations]
    end
    API <--> DB
    API <--> OLLAMA
    MIGRATIONS --> DB

API Endpoints

Method	Endpoint	Description
GET	/text	List first 20 texts
POST	/text	Add texts (array of { content })
GET	/text/search	Vector search (?query=...)

Testing the API

You can use the provided .http file to easily test the API endpoints directly from Visual Studio Code or other compatible tools. This file contains example requests for adding, searching, and listing texts.

EmbeddingPoC

A proof-of-concept .NET API for text embedding and semantic search using PostgreSQL with pgvector and Ollama for embedding generation.

Features

Store and retrieve text embeddings in PostgreSQL using Entity Framework Core and pgvector.
Generate embeddings via Ollama API.
REST endpoints for:
- Adding new texts and their embeddings.
- Searching for similar texts using vector similarity.
- Listing stored texts.

Usage

Simply run:

docker-compose up --build

…

View on GitHub

Add the MCP server to the ASP.NET Core minimal API

ohalay — Thu, 18 Sep 2025 14:37:43 +0000

A running solution where you can add server middleware to ASP.NET Core minimal APIs

1. Install MCP NuGet packages

Install the MCP packages from NuGet. Use the latest stable versions where possible. Example package links:

2. Add the MCP server to your host

Register the MCP server and transports during host building. WithToolsFromAssembly() scans your assembly for [McpServerToolType] classes and auto-registers tools.

builder.Services
    .AddMcpServer()
    .WithHttpTransport()
    .WithToolsFromAssembly();

var app = builder.Build();

app.MapEndpoints(); // Your own extension method, defined later
app.MapMcp(pattern: "api/mcp"); // Expose MCP over HTTP

await app.RunAsync();

3. Define MCP tools

Mark the types that contain tools with the appropriate attribute McpServerToolType and annotate tool methods McpServerTool. Provide clear Description text to improve discovery and documentation.

Note: attribute names and exact signatures depend on the MCP C# SDK you're using.

[McpServerToolType]
public static class Endpoint
{
    public static IEndpointRouteBuilder MapEndpoints(this IEndpointRouteBuilder app)
    {
        app.MapGet("api/users", Handler)
           .WithSummary("Get list of users")
           .Produces<PagedResponse<GetUserResponse>>(StatusCodes.Status200OK)
           .WithOpenApi();

        return app;
    }

    [McpServerTool(Name = "get_paged_list_users"), Description("Get cursor-based paged list of users")]
    private static async Task<PagedResponse<GetUserResponse>> Handler(ApplicationDbContext dbContext, [AsParameters] GetUsersRequest request, IValidator<GetUsersRequest> validator, CancellationToken token)
    {
        await validator.ValidateAndThrowAsync(request, token);

        var users = await dbContext.Users
            .Select(user => Map(user))
            .ToPagedAsync(request, token);

        return users;
    }
}

4. Add authorization

The Model Context Protocol provides authorization capabilities. In our API, we are using OAuth 2.1, and we will reuse it. Just add RequireAuthorization to the MCP endpoint.

// Add JWT authentication
builder.Services.AddAuthentication(JwtBearerDefaults.AuthenticationScheme)
    .AddJwtBearer(options =>
    {
        options.Authority = "jwt_authority";
        options.Audience = "jwt_audiences";
    });
builder.Services.AddAuthorization(c => c.DefaultPolicy = new AuthorizationPolicyBuilder().RequireAuthenticatedUser().Build());
...
app.MapMcp(pattern: "api/mcp").RequireAuthorization();
...

5. Test locally

Use the Model Context Protocol inspector app: https://modelcontextprotocol.io/legacy/tools/inspector
Use your VS Code with Copilot: https://code.visualstudio.com/docs/copilot/customization/mcp-servers
Any other tool that supports MCP

Conclusions

There’s a straightforward way to add MCP server support to an existing minimal API. We can reuse the same endpoint handlers and authorization logic, and dependency injection works as before. With this setup, the server can now expose both a REST API and MCP. The key point is to provide clear descriptions of tools and parameters to improve discovery and usability for LLMs.

Additional references

Model Context Protocol docs: https://modelcontextprotocol.io/docs/getting-started/intro
Getting-started blog post: https://devblogs.microsoft.com/dotnet/build-a-model-context-protocol-mcp-server-in-csharp/
GitHub C# SDK: https://github.com/modelcontextprotocol/csharp-sdk/blob/main/README.md

Integration Testing with .NET Aspire: Unified Local and Test environments with SQS, Lambda, and Postgres

ohalay — Tue, 24 Jun 2025 05:50:44 +0000

Our application is an ASP.NET Core API, Postage DB using Entity Framework, and SQS Lambda handler. The main idea is to configure an application using Aspire for local development, and the same configuration for integration testing.

Solution structure

ApiService - ASP.NET Core API that persists data to Postgres and sends a message to AWS SQS. EF Core code-first approach for persistent abstraction.
Lambda - AWS Lambda that is triggered based on an SQS message
AppHost - Describes and runs resources for an application.
Tests - xUnit tests that use the Aspire app host to run and test ApiService and Lambda projects.

Aspire Configuration

DB: Postgres container, PgAdmin container, and DB Migrate using ef core tool

// Postgres configuration
var postgres = builder.AddPostgres(AspireResources.Postgres)
  .WithLifetime(ContainerLifetime.Persistent)
  .WithPgAdmin(c => c.WithLifetime(ContainerLifetime.Persistent));

var db = postgres.AddDatabase(AspireResources.PostgresDb);

// DB migration
var migrator = builder
  .AddExecutable(
    "ef-db-migrations",
    "dotnet",
    "../AspirePoc.ApiService",
    "ef", "database", "update", "--no-build"
  )
  .WaitFor(db)
  .WithReference(db);

AWS: LocalStack container, Lambda emulator, Lambda handler

// LocalStack configuration
var awsConfig = builder.AddAWSSDKConfig()
  .WithRegion(RegionEndpoint.EUCentral1);

var localStack = builder.AddLocalStack(AspireResources.LocalStack)
  .WithEnvironment("AWS_DEFAULT_REGION", awsConfig.Region!.SystemName);

// AWS Lambda function configuration
builder.AddAWSLambdaFunction<Projects.AspirePoc_Lambda>(
  AspireResources.Lambda,
  "AspirePoc.Lambda::AspirePoc.Lambda.Function::FunctionHandler")
  .WithReference(awsConfig);

API: Api service with connection to DB and LocalStack

// API
var url = $"http://sqs.eu-central-1.localhost.localstack.cloud:4566/000000000000/{AspireResources.LocalStackResources.SqsName}";
var apiService = builder.AddProject<Projects.AspirePoc_ApiService>(AspireResources.Api)
  .WithReference(localStack)
  .WithReference(awsConfig)
  .WithReference(db)
  .WithEnvironment("SqsUrl", url)
  .WaitForCompletion(migrator);

Integration test implementation

xUnit fixture - Start up the Aspire test host before tests run

[assembly: AssemblyFixture(typeof(AspireFixture))]

namespace AspirePoc.Tests.Infrastructure;

public class AspireFixture : IAsyncLifetime
{
  public IDistributedApplicationTestingBuilder AppHost { get; private set; }
  public DistributedApplication? App { get; private set; }

  public async ValueTask InitializeAsync()
  {
    AppHost = await DistributedApplicationTestingBuilder
      .CreateAsync<Projects.AspirePoc_ApiService>();

    AppHost.Services.ConfigureHttpClientDefaults(clientBuilder =>
    {
      clientBuilder.AddStandardResilienceHandler();
    });

    // Remove useless resources for testing
    RemoveNotNeededResourcesForTesting();

    // Change config for testing
    ModifyResourcesForTesting();

    // Set AWS credentials for testing
    SetupTestDependencies();

    App = await AppHost.BuildAsync();
    await App.StartAsync();

    // Wait for the API service to be healthy
    using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30));
    await App.ResourceNotifications
      .WaitForResourceHealthyAsync(AspireResources.Api, cts.Token);
  }

  public async ValueTask DisposeAsync()
  {
    if (App is not null)
    {
      await App.DisposeAsync();
    }
 }

Configure test host and dependencies(Http client, DbContext, LambdaClient) to arrange tests

AppHost.Services
  .AddSingleton<IAmazonLambda>(_ => new AmazonLambdaClient(
  new BasicAWSCredentials("mock-access-key", "mock-secret-access-key"),
  new AmazonLambdaConfig
  {
    ServiceURL = AppHost.GetLambdaEmulatorUrl(),
  }))
  .AddSingleton(_ =>
  {
    var connectionString = AppHost.GetConnectionString()
      .GetAwaiter().GetResult();
    var npqConnBuilder = new NpgsqlConnectionStringBuilder(connectionString)
    {
      IncludeErrorDetail = true
    };

    var source = new NpgsqlDataSourceBuilder(npqConnBuilder.ToString())
      .Build();

    return new AppDbContext(
      new DbContextOptionsBuilder<AppDbContext>()
      .UseNpgsql(source)
      .Options);
    });

API Test example

[Fact]
public async Task Get_WeatherForecast_ShouldReturnResponse()
{
  // Arrange
  dbContext.Add(new WeatherForecast(new DateOnly(), 25, "Sunny"));
  await dbContext.SaveChangesAsync(TestContext.Current.CancellationToken);

  // Act
  var response = await sut.GetFromJsonAsync<WeatherForecast[]>(
    "/weatherforecast",
    TestContext.Current.CancellationToken);

  // Assert
  response.ShouldNotBeEmpty();
}

Lambda handler test example

[Fact]
public async Task InvokeLambda_SQSLambda_ShouldAccepted()
{
  var request = new InvokeRequest
  {
    FunctionName = AspireResources.Lambda,
    InvocationType = InvocationType.Event,
    Payload = @"{
                 ""Records"": [
                   { ""body"": ""Test message"" }
                 ]
                }"
    };
    // Act
    var response = await lambdaClient.InvokeAsync(
      request,
      TestContext.Current.CancellationToken);

    // Assert
    response.StatusCode.ShouldBe(202);
}

Conclusion

The result appears in the Aspire dashboard and GitHub

aspire

ohalay — Mon, 23 Jun 2025 09:48:32 +0000

We don't need DynamoDB when we have Aurora Postgres

ohalay — Fri, 14 Mar 2025 09:30:20 +0000

Aurora Postgres has many cool features. Let's compare them to DynamoDB and review them. While I'm writing this article, the latest version of Postgres - v17.4

Key-value and document data models - We don't need to talk about key-value. It is more interested in "how to store document data?". Postregs has a special type to sore JOSN data - JSONB. It was introduced in version 9.4 more than 10 years ago. Using EF Core 8 we don't even need to think of serialization/deserialization - just use it.
Serverless that scales to zero - AWS introduced scaling to 0 capacity with Amazon Aurora Serverless v2. It is an infrastructure-level configuration, but keep in mind a cold start during resuming.
ACID transactions - nice joke 😄
Active-active replication with global tables - here is an AWS blog post on configuring active-active replication for Postgres. It is an infrastructure-level configuration.
DynamoDB Streams as part of an event-driven architecture - postgress logical replication. A Short article about AWS RDS configuration here. The .Net Npgsql data provider has abstraction for logical replication
Secondary indexes - postgres jsonb type support Generalized Inverted Index. and can efficiently search using then.
Time to live - postgres pg_cron extension that we can use to schedule row delition. But before that, we need to configure rds
Bulk import from S3 - easy, just add permission and go on.

Finally, do you want to trigger aws lambda from RDS? No problem, we can do that.

Conclusion

Before choosing DynamoDB for your project, I recommend creating a Proof of Concept (PoC) using Aurora Postgres to validate your non-functional requirements (NFRs). With its mature feature set and broad support for complex use cases, Aurora Postgres can often be an excellent alternative for many workloads, saving you from introducing the complexity of a NoSQL database when a relational solution might meet your needs just as well.

Visualize Asp Net Core metrics with Grafana

ohalay — Mon, 03 Feb 2025 08:48:39 +0000

Spoiler: In this article, we’ll walk through how to collect and visualize metrics from an ASP.NET Core application using OpenTelemetry, Prometheus, and Grafana. We’ll configure the necessary services and see how to monitor performance metrics in a real-time dashboard.

Collect metrics with OTLP

A metric is a measurement of your application’s behavior that we aggregate over a given time period and capture at runtime. OpenTelemetry (OTLP) is an industry standard for collecting and exporting metrics. Here’s an example of how to set it up in your ASP.NET Core application metrics:

appBuilder.Services.AddOpenTelemetry()
    .ConfigureResource(builder => builder
        .AddService(serviceName: "my_service_name"))
    .WithMetrics(builder => builder
        .AddAspNetCoreInstrumentation() // metric collector
    );

Produce metrics with Prometheus

Once you've collected metrics with OpenTelemetry, you can export them using any supported OpenTelemetry exporter. To export them to Prometheus, we’ll add the .AddPrometheusExporter() method to the metric configuration. It works as a pull model, with a scrap date from the application. That means, an application should provide an endpoint with metrics for scrapping. To do that we need to add middleware app.MapPrometheusScrapingEndpoint(). Prometheus scrapes this /metrics endpoint at regular intervals to collect the application’s time-series data metrics:

# HELP http_requests_total The total number of HTTP requests.
# TYPE http_requests_total counter
http_requests_total{method="post",code="200"} 1027 1395066363000

Define Grafana charts

The dotnet team created a Grafana charts for the built-in Asp Net Core metrics. We can also search the community charts published in Grafana. We just need to import it.

Run locally

We will use docker-compose with AspNet Core, Prometheus, and Grafana containers. We download Grafana charts and add them to the container (grafana folder). We also configured the Prometheus scrab job, prometheus.yml.

services:
  httpprometheus:
    image: ${DOCKER_REGISTRY-}httpprometheus
    build:
      context: HttpPrometheus
      dockerfile: Dockerfile
    environment:
      - OTEL_RESOURCE_ATTRIBUTES=service.namespace=demo,deployment.environment=dev

  prometheus:
    image: prom/prometheus
    restart: always
    ports:
      - 9090:9090
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml:ro

  grafana: 
    image: grafana/grafana
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=P@ssw0rd
    restart: always
    ports:
      - 3000:3000
    volumes:
      - ./grafana/datasource.yml:/etc/grafana/provisioning/datasources/datasource.yml
      - ./grafana/dashboards:/etc/grafana/provisioning/dashboards

Source code

ohalay / HttpPrometheus

Vsualize Asp Net Core metrics with Promethous and Grafana

An example of Asp Net Core metric configuration using Promethous and Grafana. Also docker-compose was provided to run locally.

Requirements

.NET8
Docker

View on GitHub

Help Links

Dotnet AsyncApi specification for Avro Kafka channel

ohalay — Mon, 16 Sep 2024 07:19:44 +0000

Problem

How do we document Event-Driven Architecture (EDA) and provide discovery for events and transport?

Solution

Generate AsyncAPI specifications document - it is defacto standard for EDA as OpenAPI for HTTP API. This document may be used by a lot of tools to visualize and share specifications:

Implementation

Our ASP .Net Core application publishes events to Kafka using avro serialization protocol. There is a library that can help to build AsyncApi specification. A stable version v5.2.4(when the article was published) does not support Avro schema, but the new beta version (6.0.0-beta.97) - has implementation.

1. We will start with the high-level document

var document = new AsyncApiDocument()
{
    Info = new AsyncApiInfo
    {
        Title = "My AsyncAPI",
        Version = "0.0.1"
    }
};

2. Next, we will create components(Kafka messages with Avro schema). Also, there is an option to reference the existing Avro file schema.

document.Components = new AsyncApiComponents
{
    Messages = new Dictionary<string, AsyncApiMessage>
    {
        ["MyAvroMessage"] = new AsyncApiMessage
        {
            SchemaFormat = "application/vnd.apache.avro;version=1.9.0",
            Payload = new AsyncApiAvroSchemaPayload(new AvroRecord
            {
                Doc = "Doc for event",
                Name = "MyAvroMessage",
                Namespace = "Test.MyAvroMessage",
                Fields = [
                    new AvroField {
                        Doc = "Doc for field",
                        Name = "Id",
                        Type = AvroPrimitiveType.String,
                    }
                ]
            })
        }
    }
};

3. Reference messages to channels(Kafka topic)

document.Channels["my.kafka.topic"] = new AsyncApiChannel
{
    Subscribe = new AsyncApiOperation
    {
        Message = [
           new AsyncApiMessage
           {
               Reference = new AsyncApiReference
               {
                   Id = "MyAvroMessage",
                   Type = ReferenceType.Message,
               }
           }
        ]
    }
};

4. Serialize document to AsyncApi specification

var serializedDoc = document.Serialize(
    AsyncApiVersion.AsyncApi2_0,
    AsyncApiFormat.Yaml
);

5. In the end, we will have the next specification

asyncapi: 2.6.0
info:
  title: My AsyncAPI
  version: 0.0.1
channels:
  my.kafka.topic:
    subscribe:
      message:
        $ref: '#/components/messages/MyAvroMessage'
components:
  messages:
    MyAvroMessage:
      payload:
        type: record
        name: MyAvroMessage
        namespace: Test.MyAvroMessage
        doc: Doc for event
        fields:
          - name: Id
            type: string
            doc: Doc for field
      schemaFormat: application/vnd.apache.avro;version=1.9.0

6. Finally, visualization

Conclusion

With AsyncApi spec document we can share it via endpoint as OpenAPI document and use a tool to visualize
We can build an Avro scheme from scratch or reference schema file
We can provide information for Kafka server and bindings, but it is not our case
We can even generate clients for some programming languages, but that is also not our case
Currently AsyncAPI.NET don't support 3.0.0 specification

Help links

CloudWatch OTLP metrics exporter using .Net Lambda

ohalay — Sat, 27 Jul 2024 08:13:09 +0000

We will configure the open telemetry protocol(OTLP) metrics exporter to AWS CloudWatch using AWS Lambda with .NET 8 runtime. We will start from the infrastructure using AWS CDK.

Lambda infrastructure

To support Lambda for open telemetry, we need to configure the next thing. Add a lambda layer for AWS Distro for OpenTelemetry(ADOT); layer versions are in GitHub repository. Also, add collector.yaml path to the environment variables.

const string version = "ver-0-115-0:2";
var lambda = new Function("otlp-lambda", new FunctionProps
{
  ...
  Runtime = Runtime.DOTNET_8,
  Environment = new Dictionary<string, string>
  {
    ["OPENTELEMETRY_COLLECTOR_CONFIG_URI"] = "/var/task/collector.yaml";
     },
  Layers = [LayerVersion.FromLayerVersionArn(
     this,
     "otel-lambda-layer",
     Fn.Join("", ["arn:aws:lambda:", Aws.REGION, $":901920570463:layer:aws-otel-collector-arm64-{version}"]))
  ]
}

Collector config

The collector file is a configuration to receive, process, and export telemetry data. An important thing is NodeName that we will path later, during OTLP configuration. ADOT collector configuration here.

receivers:
  otlp:
    protocols:
      grpc:
      http:
exporters:
  debug:
    verbosity: normal
  awsemf:
    log_group_name: '/aws/lambda/{NodeName}'
service:
  pipelines:
    metrics:
      receivers: [ otlp ]
      exporters: [ awsemf ]

.NET part

First of all, we need to register open telemetry metrics that we want to collect and add an exporter. AWS_LAMBDA_FUNCTION_NAME - environment variable that is managed by AWS.

.AddOpenTelemetry()
  .WithMetrics(builder =>
  {
    var lambdaName = Environment.GetEnvironmentVariable("AWS_LAMBDA_FUNCTION_NAME")
      ?? "unknown";
    var resourceBuilder = ResourceBuilder.CreateDefault()
      .AddService("otlp-metrics-sample")
      .AddAttributes([new KeyValuePair<string, object>("NodeName", lambdaName!)]);

    builder
    .SetResourceBuilder(resourceBuilder)
    .AddMeter(api.test.operation);
    .AddOtlpExporter((_, readerOptions) =>
    {
        readerOptions.TemporalityPreference = MetricReaderTemporalityPreference.Delta;
    })
  });

Next, we need to integrate metrics with the DI container .AddMetrics().
The last thing - record our metric

app.MapPost("api/metric", (IMeterFactory meterFactory) =>
{
  var metter = meterFactory.Create("api.test.operation");
  var instrument = metter.CreateCounter<int>("sample_counter");
  instrument.Add(1);

  return Results.Ok();
});

CloudWatch

ADOT collector will create a separate stream in the AWS Lambda log group with the prefix otel-stream- and put all metrics there. After that, we can use those metrics with CloudWatch tools(dashboards, alarms, etc).

"OTelLib": "api.test.operation",
"Version": "1",
"_aws": {
  "CloudWatchMetrics": [
    {
      "Namespace": "otlp-metrics-sample",
      "Dimensions": [
        [ "OTelLib" ]
      ],
      "Metrics": [
        {
          "Name": "sample_counter"
        }
      ]
    }
  ],
  "Timestamp": 1722233676723
},
"sample_counter": 2

Conclusion

Open telemetry - language agnostic protocol, and using the built-in exporter, we may easily migrate from AWS CloudWatch to other supported exporters.
Using .NET abstraction IMeterFactory for OTLP, we may easily migrate from AWS Lambda to other computed services(Any Cloud, On-Prem, etc), with no vendor lock.

Help links

Implement REST async request-reply pattern in .Net 8

ohalay — Thu, 15 Feb 2024 11:14:28 +0000

Decouple backend processing from a frontend host, where backend processing needs to be asynchronous, but the frontend still needs a clear response. More details are in the article.

ToDo API

I'm going to show implementation based on ToDo API. We have REST methods to get ToDo(s). The typical implementation - gets ToDo item by ID and gets all ToDo items

group.MapGet("/", (Store store) => store.Todos.Values.ToArray());
group.MapGet("/{id}", (string id, Store store) =>
  store.Todos.TryGetValue(id, out var todo)
    ? Results.Ok(todo)
    : Results.NotFound());

More interested POST method, because it is asynchronous:

Accept a request and generate jobId
Create a job and send it to the queue
Return 202 accepted status code with response header Location URL where get the result and RetryAfter to delay next request.

group.MapPost("/", async (Todo model, 
   BackgroundJobQueue queue,
   Store store, HttpResponse response) =>
{
    model = model with { Id = Guid.NewGuid().ToString() };
    var job = new Job(
     Guid.NewGuid().ToString(),
     DateTime.UtcNow,
     JobStatus.InProgress,
     $"/api/todos/{model.Id}"
    );

    store.Jobs.Add(job.Id, job);

    Func<CancellationToken, ValueTask> workItem = async (token) =>
    {
      await Task.Delay(TimeSpan.FromSeconds(10), token);
      store.Todos.Add(model.Id, model);

    };
    await queue.QueueJobAsync(job.Id, workItem);

    response.Headers.RetryAfter = TimeSpan.FromSeconds(2).ToString();
    response.Headers.Location = $"/api/jobs/{job.Id}";
    response.StatusCode = StatusCodes.Status202Accepted;
});

Background service

In this example, I'm using ASP Net Core build-in BackgroundService as a job processor and Channel as an in-process queue. But it may be any job processor like hangfire, lambda, etc. Also, it may be any queue in-process or hosted. Just take into account all the pros and cons related to queues and job processors.

private async Task BackgroundProcessing(CancellationToken stoppingToken)
{
  while (!stoppingToken.IsCancellationRequested)
  {
    var (id, jobTask) = await queue.DequeueJobAsync(stoppingToken);
    if (!store.Jobs.TryGetValue(id, out var job))
      return;

    try
    {
      await jobTask(stoppingToken);
      job = job with { CompletedAt = DateTime.UtcNow, Status = JobStatus.Completed };
    }
    catch (Exception ex)
    {
      job = job with { CompletedAt = DateTime.UtcNow, Error = ex.Message, Status = JobStatus.Failed };
    }

    store.Jobs[id] = job;
  }
}

Job API

This is a job status API that redirects to Location where get the job completion results or returns the job with an error when it is failed. Also, return the job with RetryAfter header when it is still processing.

group.MapGet("/{id}", (string id, Store store, HttpResponse httpResponse) =>
{
  if (!store.Jobs.TryGetValue(id, out var job) || job is null)
    return Results.NotFound();

  var okResult = () =>
  {
    httpResponse.Headers.RetryAfter = TimeSpan.FromSeconds(5).ToString();
    return Results.Ok(job);
  };

  return job.Status switch
  {
    JobStatus.Completed => Results.Redirect(job.Location),
    JobStatus.InProgress => okResult(),
    JobStatus.Failed => Results.BadRequest(job),
    _ => throw new NotImplementedException(),
  };
});

return group;

Test using the .http file

.http files are a simple way to quickly invoke your API endpoints

Get all ToDo items
Create a new ToDo
Using Location to get job status
Get ToDo item by id

GET {{base_url}}/api/todos

###
# @name createTodo
POST {{base_url}}/api/todos
Content-Type: application/json

{
    "dueBy": "2024-02-14",
    "title": "Write an article"
}

###
@jobLocation = {{createTodo.response.headers.Location}}

# @name job
GET {{base_url}}{{jobLocation}}
Accept: application/json

###
@todoId = {{job.response.body.$.id}}

GET {{base_url}}/api/todos/{{todoId}}

Conclusion

I provided an example, of how to implement the async request-reply pattern. Important things such as queues and job processors should be chosen related to the requirements. Also, do not forget to expire job items in the DB. Btw API clients should use status codes and headers 😎. And finally, the source code in the GitHub repository

ohalay / async-request-reply

Asynchronous Request-Reply pattern implementation

Decouple backend processing from a frontend host, where backend processing needs to be asynchronous, but the frontend still needs a clear response

Requirements

.NET8

View on GitHub

Help links

GitLab feature management for .Net

ohalay — Thu, 17 Aug 2023 16:01:24 +0000

Motivation

Provide a mechanism for continued delivery together with continued development.

Solution

Use trunk-based development and feature toggles.

Implementation

Before starting to work with a feature, we need to create a new feature flag in GitLab and disable it. After that, all development should be branched with a feature toggle. When the feature is developed and deployed we can enable a feature in GitLab for the environment and it will activate during runtime. After the next release, we delete a feature flag and clean up the code.

GitLab feature management

GitLab provides a built-in solution for feature flags where we can create, add strategies, and manage. GitLab uses unleash client(we can also use direct GitLab API) to work with features. We can install unleash NuGet package from here.

During development, I found two weird things

UnleashSettings has a specific property for Environment, but the client works only when we set the GitLab environment name to AppName.
It is hard to understand where we should set Instance ID in UnleashSettings. In GitHub example - ProjectId, but property that work InstanceTag

var unleash = new DefaultUnleash(new UnleashSettings
{
   AppName = "environment_name" 
   UnleashApi = new Uri("API URL in GitLab config"),
   InstanceTag = "Instance ID in GitLab config",
   SendMetricsInterval = null,
});

Environments

Important GitLab feature - environments. We deploy our solution to specific environments and we can activate/deactivate feature flags per environment.

Microsoft feature management

Unleash client is good enough, but we want to use
dotnet feature management, which has great integration with ASP.NET Core. To do that we should implement IFeatureDefinitionProvider for GitLab

internal class GitLabFeatureProvider : IFeatureDefinitionProvider
{
  private IUnleash unleash;
  public GitLabFeatureProvider(IUnleash unleash)
    => this.unleash = unleash;

  public async IAsyncEnumerable<FeatureDefinition> GetAllFeatureDefinitionsAsync()
  {
    await Task.CompletedTask;
    foreach (var feature in this.unleash.FeatureToggles)
    {
        yield return CreateFeatureDefinition(feature.Name, feature.Enabled);
    }
  }

  public Task<FeatureDefinition> GetFeatureDefinitionAsync(string featureName)
  {
    var active = unleash.IsEnabled(featureName);
    return Task.FromResult(CreateFeatureDefinition(featureName, active));
  }

  private static FeatureDefinition CreateFeatureDefinition(string name, bool active)
  {
    return new FeatureDefinition
    {
      Name = name,
      EnabledFor = active
        ? new[] { new FeatureFilterConfiguration { Name = "AlwaysOn" } }
        : Array.Empty<FeatureFilterConfiguration>()
    };
  }
}

Using feature management

In the end, we need to register feature management

.AddFeatureManagement()
.AddSingleton<IFeatureDefinitionProvider, GitLabFeatureProvider>();

inject IFeatureManager into our solution and use it
await feature.IsEnabledAsync(nameof(FeatureFlags.myfeature))

Conclusions

We manage our features during runtime using GitLab feature flags per environment
We can rollback feature during runtime
We develop and deploy without impact to the system and end users.

Help links

AWS DynamoDB stream batch processing lambda using CDK

ohalay — Mon, 31 Jul 2023 08:13:50 +0000

Simple lambda

The most common mechanism create lambda - install AWS Toolkit, which will add AWS project templates. After that, choose the lambda template with a DynamoDB trigger, and that is all you need. But what about infrastructure?

public void FunctionHandler(DynamoDBEvent dynamoEvent)
{
  foreach (var record in dynamoEvent.Records)
  {   
    // TODO: Add business logic processing the record.Dynamodb
  }
}

CDK

Infrastructure should live together with code because it has the same lifetime. Also, infrastructure should be easily reproducible in other environments. CDK is IaC that will solve these problems and we can write infrastructure using our lovely C# language.

public class DynamoDbLambda : Stack
{
  internal DynamoDbLambda(Construct scope, string id, IStackProps props) 
     : base(scope, id, props)
  {
    var dynamoDbTable = new Table(this, "testTable", new TableProps
    {
      BillingMode = BillingMode.PAY_PER_REQUEST,
      PartitionKey = new Attribute { Name = "Pk", Type = AttributeType.STRING },
      SortKey = new Attribute { Name = "Sk", Type = AttributeType.STRING },
    });

    var dynamoDbLambda = new Function(this, "dynamoDbLambda", new FunctionProps
    {
      Runtime = Runtime.DOTNET_6,
      MemorySize = 256,
      Handler = "Serverless.DynamoDbLambda::Serverless.DynamoDbLambda.LambdaHandler::Handle",
      Code = Code.FromAsset("Serverless.DynamoDbLambda/", new AssetOptions
      {
        Bundling = buildOption
      }),
    });

    dynamoDbLambda.AddEventSource(new DynamoEventSource(dynamoDbTable, new DynamoEventSourceProps
    {
      StartingPosition = StartingPosition.TRIM_HORIZON,
    }));
  }
}

Taking into account, that we are using DynamoDB single table design, we will receive stream for all tables, but we want to listen to only а special table...

Filters

AWS lambda source has the possibility filter Dynamo DB stream. CDK syntax is a little bit weird, but it is just JSON representations.

Filters = new[]
{
  FilterCriteria.Filter(new Dictionary<string, object> {
    ["dynamodb"] = new Dictionary<string, object>
    {
      ["Keys"] = new Dictionary<string, object>
      {
        ["Pk"] = new Dictionary<string, object>
        {
          ["S"] = new[]{"prefix","MyTableName" }
        }
      }
    }
  })
},

All good, only one thing... handle failed records.

Fails and retries

There's no easy way to handle failed records. Because we can have a lot of reasons for failures:

Bugs in lambda code
Unavailable services
Db structure changed We may retry and in case of failure send failed record to SQS.

var deadLetterQueue = new Queue(this, "deadLetterQueue");
dynamoDbLambda.AddEventSource(
  new DynamoEventSource(dynamoDbTable, new DynamoEventSourceProps
  {
    OnFailure = new SqsDlq(deadLetterQueue),
    RetryAttempts = 5,
  }));

So far so good, but how to work with partial batch failure?

Batch processing

Instead of returning void from the lambda handler, we can report failed items and lambda will handle partial failure. Add this option to CDK ReportBatchItemFailures = true and modify the lambda itself.

public async Task<StreamsEventResponse> Handle(DynamoDBEvent events)
{
  var tasks = events.Records
   .Select(new Executor().Execute)
   .ToList();

  var failedItems = new List<BatchItemFailure>();

  try
  {
    await Task.WhenAll(tasks);
  }
  catch
  {
    failedItems = tasks
      .Select((task, index) => new { task, index })
      .Where(x => !x.task.IsCompletedSuccessfully)
      .Select(x => new BatchItemFailure 
      {
        ItemIdentifier = events.Records[x.index].EventID 
      })
      .ToList();
  }

  return new StreamsEventResponse {BatchItemFailures = failedItems};
}

And finally, the source code in the GitHub repository add-dynamo-db-lambda branch

ohalay / serverless-containers

Serverless integration tests

A public feed with available products that updates every day

Business problem

Integration tests for serverless solution

Requirements

Docker

Implementation

S3 public buckets with available documents
Lambda updates document

View on GitHub