In Part 1 and Part 2 of this tutorial series, you set up an LLM and an MCP tool for Linear in Confluent Cloud, and then built a Streaming Agent that turns customer communications into Linear issues. The agent can see what's already in Linear, but it knows nothing about the customer who wrote in. A message like "I would like a partial refund on order 9189" becomes an issue that a person has to look up, finding the order and the customer's membership level, before deciding how urgent it is.
In this tutorial, you'll give the agent that context with the Real-Time Context Engine (RTCE). RTCE materializes Kafka topic data into a low-latency serving layer and exposes it to agents as MCP tools. The agent can query current data without a Kafka consumer, a separate database, or a Flink job to maintain. You'll:
A note on how this tutorial relates to a broader Streamhouse data architecture: Streamhouse is an open, vendor-neutral category of data architectures that keep the current state of a business continuously available to production applications, analytics, and AI agents. It describes five stages: capture, transport, transform, govern, and serve. In this series, customer communications and orders are records in Kafka topics (transport), and Flink SQL invokes the agent on each new communication (transform). The data is schematized throughout: Flink registers a schema in Schema Registry for each table's topic, and RTCE requires a registered schema in order to expose a topic, so the agent queries typed, governed columns (govern). RTCE is the serve stage: the current state of each order is queryable by the agent at the moment it handles a communication, with no batch export in between.
Ensure that the agentic-ai-env environment is active:
confluent environment listIf it isn't, set it to be the active environment:
confluent environment use <ENVIRONMENT_ID>Start a Flink SQL shell:
confluent flink shell --compute-pool \
$(confluent flink compute-pool list -o json | jq -r ".[0].id")Create an orders table. Each order records the customer and their membership level at the time of the order. The primary key, upsert changelog mode, and compacted cleanup policy make RTCE keep only the latest record for each order.
RTCE's upsert requirements say that the Kafka record key must be raw bytes. Structured key schemas (Avro, Protobuf) aren't supported. When a table has a primary key, Flink defaults to the avro-registry key format, so set 'key.format' = 'raw' explicitly. A raw key must be a single column named key, so the order number goes in a column with that name.
CREATE TABLE orders (
`key` STRING,
customer_name STRING,
membership_level STRING,
order_total DECIMAL(10, 2),
order_status STRING,
PRIMARY KEY (`key`) NOT ENFORCED
) WITH (
'changelog.mode' = 'upsert',
'kafka.cleanup-policy' = 'compact',
'key.format' = 'raw'
);Creating the table also creates a Kafka topic named orders and registers its value schema in Schema Registry. RTCE requires a registered schema.
Insert a few orders. In production, an order management system would produce these, or a CDC connector would capture them from its database. Order 9189 is the one from the refund request in Part 2:
INSERT INTO orders
VALUES
('9189', 'Alex Rivera', 'GOLD', 214.50, 'SHIPPED'),
('9190', 'Jordan Lee', 'PLATINUM', 489.00, 'SHIPPED'),
('9191', 'Sam Patel', 'STANDARD', 36.25, 'PROCESSING');Confirm that the orders are there:
SELECT * FROM orders;Exit the Flink SQL shell (exit;).
Enable RTCE on the orders topic. Set --region to your cluster's region. The cloud must be aws:
confluent rtce rtce-topic create \
--cloud aws \
--region us-east-1 \
--topic-name orders \
--description "Customer orders. The key column holds the order number. Each order has the customer's membership_level (PLATINUM, GOLD, or STANDARD)."Agents see the description when they discover topics, so describe what the topic contains and what its key column means.
Check the phase of the RTCE topic:
confluent rtce rtce-topic describe ordersRun the command every few seconds until Phase changes from PENDING (or PROVISIONING) to ACTIVE. Don't continue until it's ACTIVE.
RTCE reads the topic's cleanup policy to decide how to materialize it. The orders topic uses the compact policy and a raw key, so RTCE materializes it in upsert mode and keeps only the latest record for each order number (the key column).
RTCE's MCP endpoint authenticates with a Confluent Cloud API key that has global scope. Create one:
confluent api-key create --resource global --description "RTCE tutorial"Save the API key and secret. You'll need both to create a Flink connection in the next step.
Next, get the MCP endpoint URL. In the Confluent Cloud Console, navigate to your cluster and click Topics. In the orders row, find the Context engine column, click Details, and then click Copy topic details to clipboard. The endpoint resembles:
https://mcp.{CLUSTER_REGION}.aws.confluent.cloud/mcp/v1/context-engine/organizations/{ORG_ID}/environments/{ENV_ID}/kafka-clusters/{LKC_ID}Start a Flink SQL shell again:
confluent flink shell --compute-pool \
$(confluent flink compute-pool list -o json | jq -r ".[0].id")Create a connection to RTCE's MCP server. The Linear connection in Part 1 used a bearer token, but RTCE uses HTTP Basic authentication. Use the API key as the username and the API secret as the password:
CREATE CONNECTION `rtce-mcp-connection`
WITH (
'type' = 'mcp_server',
'endpoint' = '<RTCE_MCP_ENDPOINT>',
'transport-type' = 'STREAMABLE_HTTP',
'username' = '<API_KEY>',
'password' = '<API_SECRET>'
);RTCE exposes three MCP tools:
Create a tool that lets the agent call all three:
CREATE TOOL rtce_tool
USING CONNECTION `rtce-mcp-connection`
WITH (
'type' = 'mcp',
'allowed_tools' = 'listTopics,getMetadata,queryData',
'request_timeout' = '30'
);Add the RTCE tool to the chat_listener_agent from Part 2. The agent keeps its Linear tool and now has both:
ALTER AGENT chat_listener_agent ADD TOOL rtce_tool;The agent now makes more tool calls for each communication: it looks up the order, checks Linear for similar issues, and creates the issue. Raise the iteration limit from 5 in Part 2 so the agent has room to finish:
ALTER AGENT chat_listener_agent SET ('max_iterations' = '10');Next, update the prompt. In addition to the Part 2 instructions, it tells the agent to look up the order in RTCE and set the Linear issue priority from the customer's membership level. In Linear, 1 is Urgent, 2 is High, 3 is Medium, and 4 is Low. The prompt names the topic and columns so the agent doesn't have to discover them on every run:
ALTER AGENT chat_listener_agent
SET PROMPT 'Create an issue from the content using <LINEAR_TEAM_ID> as the team ID. Do not create a new issue if a similar issue already exists. If the content mentions an order number, look up that order in the orders topic using the Real-Time Context Engine tool, querying by the key column, and read its membership_level. Use the membership_level to set the issue priority: PLATINUM is priority 1, GOLD is priority 2, and STANDARD is priority 3. If no order is mentioned or the order is not found, use priority 4. Mention the order number and membership level in the issue description.';Insert two new customer communications that refer to the new orders:
INSERT INTO customer_communications
VALUES
(UUID(), 'Hi, order 9190 arrived with a cracked screen on the display unit. I would like a replacement sent out.'),
(UUID(), 'Hello, my order 9191 has been processing for days and I would like to know when it will ship.');Run the agent:
SELECT
status,
response
FROM customer_communications,
LATERAL TABLE(AI_RUN_AGENT('chat_listener_agent', `prompt`, `id`, MAP['debug', true]));This query processes every row in customer_communications, including the ones from Part 2. The agent sees that issues for those already exist and doesn't create new ones. With debug enabled, you can see the agent call the RTCE tool to look up the order.
Open your Linear workspace to see the new issues. Order 9190 belongs to a PLATINUM customer, so its issue should have Urgent priority. Order 9191 belongs to a STANDARD customer, so its issue should have Medium priority. LLM output varies from run to run, so the wording and tool calls may differ, but the priorities should follow the membership levels.
So far, you've run the agent with a SELECT query. To run it continuously, so it handles every new customer communication as it arrives, write the results to a table with an INSERT INTO statement:
CREATE TABLE agent_results (
id STRING,
status STRING,
response STRING
);INSERT INTO agent_results
SELECT
id,
status,
response
FROM customer_communications,
LATERAL TABLE(AI_RUN_AGENT('chat_listener_agent', `prompt`, `id`));This statement keeps running. Insert another communication from a second Flink SQL shell and watch a new issue appear in Linear:
INSERT INTO customer_communications
VALUES
(UUID(), 'Hi, I was charged twice for order 9189. Can you please refund the duplicate charge?');Try extending the agent:
When you're done experimenting, delete the agentic-ai-env environment to clean up the Confluent Cloud infrastructure created for this tutorial. Run the following command in your terminal to get the environment ID of the form env-123456 corresponding to the environment named agentic-ai-env:
confluent environment listDelete the environment:
confluent environment delete <ENVIRONMENT_ID>Also delete the global API key you created for RTCE if you no longer need it:
confluent api-key delete <API_KEY>