An AI agent is only as good as the context it reads and only as safe as the actions it is allowed to take. Nightly extracts give agents yesterday’s truth and no audit trail. The pattern we build instead puts Kafka in the middle: change data capture streams every system of record into governed topics, Flink turns those streams into entity-centric materialized context, a low-latency serving layer answers keyed lookups in milliseconds, and an MCP server exposes that context to agents as tools and resources. Every action an agent takes is written back to Kafka as an immutable action log, so you can audit it, replay it, and fork the stream to test a new agent version against real traffic without touching production. Below are the reference architecture, the patterns that make it work, a claims-agent walkthrough, a decision table, the failure modes we plan for, and a readiness checklist.
Why agents need streams, not snapshots
Most first-generation agent projects connect a model to a vector index built from a nightly export and a handful of REST APIs. That works for a demo. It breaks in production for three reasons that have nothing to do with the model.
- Staleness. An agent that answers a customer from a snapshot taken at 2 a.m. will confidently tell them their payment has not arrived when it cleared an hour ago. The model is right about the data it was given; the data is wrong.
- Scatter. The facts an agent needs about one customer, claim or order live in five systems. Calling five APIs on every turn is slow, brittle under load, and pushes join logic into prompts where nobody can test it.
- No memory of what it did. When an agent issues a refund, changes a reserve or sends a letter, that action needs to be recorded with the context it acted on and the authority it acted under. A log line in an application server is not an audit trail.
Event streaming solves all three because it was built for exactly this shape of problem: many sources, many consumers, ordered changes per key, durable history and replay. What changes with agents is not the platform but who is consuming it — and that consumer needs context served in a form it can reason over, with governance it cannot get around.
The shift in one sentence: stop asking the agent to go and find the facts, and start keeping the facts it needs continuously assembled, fresh and governed, one lookup away.
The reference architecture
The architecture has five layers and one loop. Data flows left to right from systems of record to the agent. Actions flow back from the agent into Kafka, where they become events like any other — consumed by downstream systems, auditors and, if you want, the next version of the agent.
Layer 1 — Capture changes, not copies
Debezium reads the transaction log of each operational database and emits one event per row change, in commit order, per key. Kafka Connect brings in SaaS systems, files and message queues. The rule is that nothing is extracted on a schedule if it can be captured as it changes: the agent’s freshness can never be better than the freshness of its slowest source.
Layer 2 — Governed topics
Every topic has an owner, a schema registered with compatibility rules, a retention policy and ACLs. Raw CDC topics are treated as internal: they carry every column the source has, including the ones an agent should never see. Agents and their tools never read raw topics directly.
Layer 3 — Flink builds the context
Flink joins the streams that describe one business entity — a customer, a policy, a claim, an order — into a single, continuously updated record. It enriches it with reference data, computes the aggregates an agent would otherwise ask for (open balance, last contact, claim count in 12 months), and masks or tokenizes sensitive fields before anything leaves the stream. The output is a small number of curated, agent-ready topics.
Layer 4 — A serving layer for lookups
Agents ask keyed questions: “everything about claim 4471.” Kafka is not a database and should not be asked to answer that by scanning a topic. The curated topics are sunk into a key-value store for entity context, a feature store for scores and aggregates, and a vector index whose embeddings are refreshed when the underlying record changes rather than once a night.
Layer 5 — MCP as the contract with the agent
The Model Context Protocol gives agents a standard way to discover and call tools and read resources. We expose the serving layer through an MCP server with a small, deliberately designed tool surface: typed lookups (get_claim_context), bounded searches (find_similar_claims, top-k with filters), and action tools that do not call downstream systems directly but write a command event to Kafka. The MCP server is where identity, authorization, rate limits and logging are enforced — once, for every agent.
Seven patterns that make it work
Entity-centric materialized context
Design the context around the nouns the agent reasons about, not around source tables. One record per claim, versioned, with a context_version and an as_of timestamp. The agent reads one record instead of assembling ten, and every answer can cite the exact version it used.
Keyed and filtered reads, never topic scans
An agent that needs the last three events for one customer should never consume a partition from the beginning to find them. Key topics by entity ID so related events stay ordered on one partition, materialize what is read often, and push filters to the serving layer. Topic scans for agent context are how you get unpredictable latency and surprise egress bills.
Freshness as a measured SLO
Define freshness per context type — for example, payments within seconds, documents within minutes — and measure it end to end: source commit time to serving-layer write time. Expose as_of to the agent so it can say “as of 10:42” and, when freshness is breached, degrade gracefully instead of answering from stale data.
Idempotent, exactly-once actions
Agents retry. Models time out. Users click twice. Every action tool takes an idempotency key derived from the conversation, the entity and the intent, and writes a command event using Kafka’s idempotent producer and transactions. The consumer that executes the command deduplicates on that key. A refund requested three times is issued once.
The action log is the source of truth for agent behavior
Each action event carries the agent identity and version, the tool, the inputs, the context version it read, the policy that authorized it, and the outcome. That record is immutable and retained long enough for your regulators. It is what you show an auditor, what you replay to debug, and what you feed back into evaluation.
Fork the stream to test agents safely
Because context and actions are both streams, a new agent version can run in shadow mode: it consumes the same curated topics, makes its decisions, and writes them to a separate agent.actions.shadow topic that no downstream system executes. Compare shadow and production decisions on real traffic before promoting. Replaying a past window of events reproduces an incident exactly.
Least privilege for non-human identities
Each agent gets its own service identity with ACLs scoped to the curated topics and tools it needs — read on claims.context, write on claims.commands, nothing on raw CDC. Authority limits (maximum payment, which letters it may send) are enforced in the MCP server and again in the command consumer, so a prompt injection cannot talk its way past them.
What the context materialization looks like
Most of the logic lives in Flink SQL. This is a simplified version of the job that builds claim context for a claims agent: one row per claim, refreshed on every change, with the sensitive fields masked before the record leaves the stream.
-- Curated, agent-ready context: one row per claim, upserted on every change CREATE TABLE claims_context ( claim_id STRING, policy_id STRING, status STRING, loss_date DATE, reserve_amount DECIMAL(14,2), paid_to_date DECIMAL(14,2), open_tasks INT, last_contact_at TIMESTAMP(3), claimant_name STRING, -- masked below as_of TIMESTAMP(3), PRIMARY KEY (claim_id) NOT ENFORCED ) WITH ('connector' = 'upsert-kafka', 'topic' = 'claims.context', ...); INSERT INTO claims_context SELECT c.claim_id, c.policy_id, c.status, c.loss_date, r.reserve_amount, p.paid_to_date, t.open_tasks, t.last_contact_at, CONCAT(SUBSTRING(c.claimant_first, 1, 1), '. ', c.claimant_last) AS claimant_name, GREATEST(c.updated_at, r.updated_at, p.updated_at, t.updated_at) AS as_of FROM cdc_claims c LEFT JOIN latest_reserves r ON r.claim_id = c.claim_id LEFT JOIN payments_rollup p ON p.claim_id = c.claim_id LEFT JOIN task_rollup t ON t.claim_id = c.claim_id;
The MCP tool that serves it is deliberately narrow. It reads one record by key from the serving store, checks that the calling agent is entitled to that line of business, and returns the record with its as_of so the agent can cite it:
// MCP tool definition (abridged) { "name": "get_claim_context", "description": "Current governed context for one claim, with as_of timestamp.", "inputSchema": { "type": "object", "properties": { "claim_id": { "type": "string", "pattern": "^CL-[0-9]{6,}$" } }, "required": ["claim_id"] } }
The request path and its latency budget
When an agent turn is interactive, every hop has to earn its place. The serving layer exists so that the hot path never touches Kafka, Flink or the source databases — they keep the context fresh in the background, and the agent reads a finished record.
as_of.Walkthrough: a claims agent on live context
Here is how the pieces work together for a claims-support agent that answers adjusters’ and policyholders’ questions and can take a small number of bounded actions.
A payment clears
The payments system commits a row. Debezium emits a change event to cdc.payments within moments of the commit. No batch job is waiting for midnight.
Context updates itself
The Flink job recomputes paid_to_date for that claim, masks the payee’s bank details, and upserts a new version of the claim record to claims.context. A sink writes it to the key-value store; the vector index re-embeds the claim summary because its content changed.
The adjuster asks
“Has the first payment on CL-204471 gone out, and is anything still open?” The agent calls get_claim_context. The MCP server checks the agent’s identity and the adjuster’s line-of-business entitlement, reads one record, and returns it with as_of.
The agent answers and cites
“Yes — the first payment cleared at 10:41. Two tasks remain open: the repair estimate and the subrogation review. (Context as of 10:42.)” Every figure traces to a context version.
The agent acts within its authority
The adjuster asks it to send the policyholder a status letter. The agent calls send_status_letter, which writes a command event with an idempotency key to claims.commands. The letter service consumes it, checks the template is one the agent may send, sends it once, and emits letter.sent. The full chain — question, context version, tool call, authorization, outcome — is on agent.actions.
Next week, a new agent version
Version 4 runs in shadow on the same context for a week. Its decisions are diffed against version 3’s. Disagreements are reviewed by the claims team before promotion. Nothing version 4 decided in shadow was ever executed.
Where should the agent read from?
Not every question belongs on Kafka, and not every question belongs in a serving store. This is the decision table we apply per tool.
| Question shape | Read from | Why |
|---|---|---|
| Current state of one entity | Serving store (key-value) | Millisecond keyed lookup of the materialized record |
| “Similar” or semantic lookup | Vector index, refreshed from the stream | Top-k with metadata filters; embeddings stay current |
| Scores, aggregates, rolling counts | Feature store fed by Flink | Same values used by models and agents |
| “What just happened” to one key | Recent events, keyed and filtered | Ordered history for one entity without a full scan |
| Trigger: react when something changes | Kafka subscription | Event-driven agents start on the event, not on a poll |
| Historical analysis across many entities | Lakehouse (Iceberg / Delta) | Analytical scans belong in the lakehouse, not in an agent turn |
| Anything from a raw CDC topic | Nowhere — curate first | Raw topics carry fields agents must not see |
A useful rule: if a tool would need to read more than a handful of events to answer, it should be reading a materialized view that Flink keeps up to date instead.
The failure modes we plan for
1. Agents reading raw CDC topics. It is the fastest prototype and the most common leak. Raw change events contain every column, including the ones masked everywhere else. Agents read curated topics through the MCP server, full stop.
2. Scanning topics on the hot path. A tool that consumes from offset zero to find one customer’s events works in a test with a thousand messages and falls over at a hundred million. Materialize, key and filter.
3. Freshness nobody measures. The pipeline lags after a schema change, the serving store quietly goes stale, and the agent keeps answering confidently. Measure freshness end to end, alert on it, and tell the agent when it is breached.
4. Actions that call systems directly. When an agent’s tool calls a payment API itself, there is no idempotency, no central authority check and no record. Actions become command events; executors deduplicate and enforce limits.
5. One shared identity for every agent. If all agents run as one service account, you cannot tell which one did what, and a compromise of one is a compromise of all. One identity per agent, scoped ACLs, authority limits checked twice.
6. Schema changes that break context silently. A renamed source column turns a field to null in the context record and the agent starts saying “no payments found.” Registry compatibility rules and data contracts on curated topics catch this before production.
Readiness checklist
| Area | Ready when |
|---|---|
| Sources | Each system of record the agent depends on is captured by CDC or a connector, not a scheduled extract |
| Topics | Every topic has an owner, a registered schema, compatibility rules, retention and ACLs |
| Context model | Agent-facing entities are defined, with context_version and as_of |
| Masking | PII and payment data are masked or tokenized in Flink before curated topics |
| Serving | Keyed lookups meet the latency budget at peak load |
| Freshness | An end-to-end freshness SLO is defined, measured and alerted on per context type |
| MCP surface | Tools are typed, bounded and documented; no tool exposes raw topics or free-form queries |
| Actions | Every action is a command event with an idempotency key and an authority check in the executor |
| Identity | One service identity per agent with least-privilege ACLs |
| Action log | Immutable, retained to your compliance window, queryable for audit |
| Testing | Shadow mode and replay are available before any agent version is promoted |
The bottom line
The hard part of production agents is rarely the model. It is giving the model facts that are current, assembled and governed, and making every action it takes accountable. Kafka, CDC and Flink were already the right tools for keeping many systems in sync in real time. Put a serving layer and an MCP server in front of them, write every agent action back as an event, and the same platform becomes the context layer and the flight recorder for your agents.
Teams that start this way find that the second and third agent are much cheaper than the first: the context is already flowing, the tools already exist, and the audit trail is already there. Teams that start with nightly exports and direct API calls usually rebuild it later, under production pressure.