How I Built a Secure AI Database Gateway in Go Using MCP, Anthropic, and Kafka

taofit — Sat, 15 Aug 2026 21:37:25 +0000

Built DynamoDB Sage: a Go MCP server that lets LLM agents safely query and manage Amazon DynamoDB via natural language
A custom RiskAnalyzer validates every mutating or heavy tool call before it touches AWS — blocking destructive operations (mass writes, table drops, expensive scans) against protected tables
Kafka decouples large write operations from the live chat, so DynamoDB throttling never freezes the user experience
Prometheus provides observability: tool and DynamoDB latency, consumed capacity, Kafka lag, and security events in real time

The hype around AI "wrapper" apps is officially over. In 2026, the real engineering challenge has shifted from basic prompt engineering to a much harder problem: how do we let autonomous LLM agents interact with live, production cloud databases securely, reliably, and at scale?

When you build a system that allows users to query and manage an enterprise database like Amazon DynamoDB using natural language in a chat interface, you hit an immediate bottleneck. Giving an LLM direct execution pathways to your database is a security nightmare. Without strict guardrails, you expose your infrastructure to prompt injections, catastrophic data deletion, resource starvation, and unpredictable token budgeting.

To solve this, I built DynamoDB Sage—a production-grade, conversational data engine written in Go.

Here is a deep dive into how I used the Model Context Protocol (MCP), the Anthropic SDK, Apache Kafka, and Prometheus to build a secure, role-aware AI gateway that turns conversational text into safe, asynchronous database operations.

The Core Architecture: Natural Language to Secure Data Retrieval

The user-facing side of the platform is a real-time conversational chat interface. Under the hood, when a user asks, "Show me all high-value customer accounts updated in the last 24 hours," the application routes the request through a multi-layered pipeline:

The Language Layer: The chat box utilizes the Anthropic SDK to leverage Claude's advanced reasoning capabilities, mapping the user's intent to specific database tool schemas.
The Protocol Layer: The backend is built as an MCP (Model Context Protocol) server. MCP acts as the open standard that cleanly bridges the LLM's cognitive reasoning with our underlying Go environment.

The whole system is driven from a web dashboard built with Next.js — a real-time chat pane streaming LLM responses over SSE, plus dedicated views for tables, tools, and Prometheus-backed monitoring. A one-click "Try the live demo" mode lets anyone explore read-only against the public instance without setting up credentials.

However, letting the LLM talk directly to the database is where most architectures fail production readiness. That brings us to the core security mechanism.

1. The Security Layer: The RiskAnalyzer Interceptor

In a production environment, you cannot trust the output of an LLM blindly. Code injection or adversarial prompt attacks can trick an agent into generating malicious payloads.

To enforce defense-in-depth, I engineered a custom Go layer called the RiskAnalyzer. Every mutating or heavy JSON-RPC tool-call generated by the LLM must pass through it before it ever touches AWS. Read-only operations skip risk analysis to keep the chat feel fast.

// Simplified conceptual look at the execution barrier
func (ra *RiskAnalyzer) Analyze(ctx context.Context, req *mcp.CallToolRequest) (Assessment, error) {
    // 1. Validate structural integrity against explicit JSON schemas
    if err := ra.validateSchema(req); err != nil {
        return Assessment{}, fmt.Errorf("validation violation: structural mismatch: %w", err)
    }

    // 2. Enforce data-boundary restrictions (protected tables, read-only tables, batch-size caps)
    if err := ra.checkTableProtection(req); err != nil {
        return Assessment{}, fmt.Errorf("authorization violation: execution path blocked")
    }

    // 3. Estimate blast radius — PII fields present, capacity/RCU cost, batch size
    assessment := ra.estimateImpact(req)

    return assessment, nil
}

Why this matters for compliance: By sitting inline as a hard gateway boundary, the RiskAnalyzer ensures that even if Claude is tricked by a user into executing a destructive query, the Go layer drops the request before it reaches AWS. This architecture matches strict corporate compliance models, ensuring data safety without adding latency to read-only traffic.

2. The Scale Layer: Asynchronous Mutations via Apache Kafka

Chat systems must feel instantaneous. If a user utilizes the chat interface to execute complex database mutations (like updating thousands of customer records via an agent), performing synchronous writes directly to DynamoDB causes massive HTTP blocking, spikes in API costs, and a terrible user experience.

To solve this, I decoupled the read/write pathways using an event-driven, asynchronous pipeline powered by a distributed Apache Kafka architecture.

Reads (Synchronous & Fast): Safe query paths fetch data instantly to render back into the conversational chat UI.
Writes (Two paths): Regular mutations (put/update/delete single items) execute synchronously after risk analysis and an explicit user confirmation step. Large operations — batch writes, batch deletes, table creation, document ingestion — are serialized and published as events to a Kafka topic, then processed asynchronously by a worker pool so they never block the chat.

// Simplified producer: publish a validated mutation instead of writing directly
func (p *MutationProducer) Publish(ctx context.Context, mutation *db.Mutation) error {
    payload, err := json.Marshal(mutation)
    if err != nil {
        return fmt.Errorf("serialize mutation: %w", err)
    }

    msg := &kafka.Message{
        Topic: "dynamodb-mutations",
        Key:   []byte(mutation.TableName),
        Value: payload,
    }

    return p.writer.WriteMessages(ctx, msg)
}

// Simplified consumer group setup using Sarama
func (c *saramaConsumer) Start() error {
    ctx, cancel := context.WithCancel(context.Background())
    c.runCancel = cancel

    config := sarama.NewConfig()
    config.Consumer.Return.Errors = true
    config.Consumer.Offsets.Initial = sarama.OffsetNewest
    config.Consumer.Group.Session.Timeout = 30 * time.Second // default is 10s
    config.Consumer.Group.Heartbeat.Interval = 3 * time.Second

    group, err := sarama.NewConsumerGroup(c.brokers, c.consumerGroupName, config)
    if err != nil {
        return err
    }
    c.consumerGroup = group

    // Consume runs a session loop; reconnect on error rather than failing hard
    c.wg.Add(1)
    go func() {
        defer c.wg.Done()
        for {
            if err := c.consumerGroup.Consume(ctx, c.topics, c); err != nil {
                log.Printf("consumer error: %v", err)
                time.Sleep(5 * time.Second)
            }
            if ctx.Err() != nil {
                return
            }
        }
    }()

    select {
    case <-c.ready:
        return nil
    case <-time.After(30 * time.Second):
        return fmt.Errorf("timeout waiting for consumer to be ready")
    }
}

// ConsumeClaim hands each message off to the worker pool for concurrent execution
func (c *saramaConsumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
    for msg := range claim.Messages() {
        var mutation db.Mutation
        if err := json.Unmarshal(msg.Value, &mutation); err == nil {
            c.jobs <- &mutation // dispatched to the worker pool; workers call executor.Apply with retries
        }
        session.MarkMessage(msg, "")
    }
    return nil
}

Kafka is the primary path, consumed via Sarama's consumer group API. The Consume call runs a session loop that automatically rejoins on rebalance or transient error — if it errors out, the goroutine backs off for 5 seconds and retries rather than crashing the consumer. Each ConsumeClaim callback hands incoming messages off to a fixed-size Go worker pool over a channel, which executes writes against DynamoDB concurrently. If DynamoDB hits a temporary throttling event (such as running out of Provisioned Write Capacity Units), Kafka safely buffers the backlog on the topic while the workers retry — no data loss, no blocked chat.

As a resilience measure, the same worker-pool executor also has a direct-dispatch fallback path that bypasses Kafka entirely if the Kafka cluster itself becomes unavailable — mutations get queued in-memory and processed by the same pool, trading Kafka's durability guarantees for continued availability during an outage. Kafka remains the default and preferred path; the fallback exists purely so the system degrades gracefully rather than failing closed.

3. Observability: Operational Excellence via Prometheus

An AI system running blindly in production is a liability. You need to know exactly how much the agent costs, how fast it responds, and when it fails.

I integrated Prometheus metrics directly into the Go tool-execution layer to provide immediate, real-time telemetry:

Latency Tracking: Per-tool and per-DynamoDB-operation latency histograms, so slow queries or a degraded analyzer are visible instantly.
Capacity & Cost Signals: Consumed read/write capacity per DynamoDB operation, plus Kafka send volume and consumer lag — the numbers that map directly to AWS cost.
Security Telemetry: A counter for risk-analyzer blocks. A sudden spike in blocked payloads signals a coordinated attack or an algorithmic breakdown in tool-calling structure, and PII-detection events are tracked separately.

These metrics feed straight into standard dashboards, transforming an ambiguous AI chat application into a fully observable, predictable enterprise service.

Lessons Learned & What's Next

Building this solo surfaced a few things I'd flag for anyone building similar agentic-database systems:

Validation logic needs to evolve as fast as new attack surfaces appear. The RiskAnalyzer's schema, table-protection, and PII rules aren't a "set and forget" layer — treat it like a living security control, not a static gate.
Async-by-default for large writes is worth the added complexity the moment you're dealing with any non-trivial mutation volume — the Kafka buffering layer paid for itself the first time I simulated throttled write capacity.
Next up: extending the observability layer with distributed tracing (OpenTelemetry) across the full request path, and moving from the current global rate limiter to per-user-session rate limiting on top of the existing security telemetry.

Conclusion: The Blueprint for Production AI

Building an AI conversational application that connects to cloud data isn't just about making a cool UI. It requires applying classic, hardened software engineering practices to a non-deterministic technology.

By wrapping the Anthropic SDK and MCP inside a defensive Go framework, gating it with a role-based risk-analysis layer, scaling it via Kafka, and measuring it through Prometheus, we turn an AI chat interface into an enterprise-grade platform feature.

The code for DynamoDB Sage is available on GitHub. I'd love to know how you are solving security and database isolation challenges in your own agentic workflows!

👉 Try the live interactive demo here: dynamodb-sage.hzcentre.com

👉 Check out the GitHub Repository: github.com/taofit/MCP-Server-Dynamodb-sage

Golang #AI #ModelContextProtocol #ApacheKafka #AWS #security #CloudArchitecture #SoftwareEngineering

DEV Community: taofit