For years, we've interacted with chatbots. They answer questions, fetch information, and follow commands. But they are fundamentally reactive. They wait for us to ask. The next leap in AI is the agentic AI—a system that can reason, plan, and proactively take action to achieve a goal.

Imagine an AI that doesn't just answer "Where is my order?" but notices you're struggling with the checkout process and proactively offers help. That's not a chatbot; that's an agent.

The secret to this proactive ability isn't just a smarter model; it's real-time context. An agent is only as good as its awareness of the present moment. In this post, we'll build a fully functional pipeline for a proactive customer service agent that knows what a user is doing, as they are doing it. We'll use Java for our applications, Kafka for event streaming, and the powerful real-time processing of DeltaStream to serve as the agent's central nervous system.

The Architecture: Our Agent's Senses

Before we dive into the code, let's look at our plan. Our system has four key components:

  1. The Data Generator (Java & Kafka): A simulator that acts like a real e-commerce website, generating a constant stream of user events (page views, cart updates, chat messages) into Kafka.
  2. The Real-Time Brain Lobe (DeltaStream): This is the core of our system. DeltaStream will ingest the raw event streams from Kafka, process them in real-time, and shape them into a high-signal, contextual profile of each user.
  3. The Context Hotline (MCP Server): A simple Java web server that acts as our "Model Context Protocol" (MCP) endpoint. It provides a clean, on-demand API for the AI agent to get the real-time context from DeltaStream.
  4. The Agent (Conceptual): The final piece. The agentic AI application will call our MCP Server to get the context it needs to make intelligent, proactive decisions.

Ready? Let's build.

Part 1: The "Real World" - Simulating User Data with Java & Kafka

First, our agent needs a world to observe. We'll create a simple Java application that continuously generates user events and sends them to three different Kafka topics: page_views, cart_updates, and chat_messages. For brevity, we skip presenting the code here, but you can get the details in DeltaStream’s public examples github repo(https://github.com/deltastreaminc/examples/tree/main/ProactiveCustomerSupportAgent).

Part 2: The Sixth Sense - Real-Time Context with DeltaStream

An agent can't act on raw data. It needs insight. We'll use DeltaStream to consume the raw Kafka events and forge them into a clean, contextualized profile for each user. This is the most critical piece of our architecture.

In your DeltaStream workspace, run the following SQL statements.

Step 1: Create Streams on Kafka Topics First, we tell DeltaStream about our Kafka topics, creating a live, queryable STREAM for each one.

  1. -- Step 1: Create Streams for each Kafka topic.
  2. -- A stream is a real-time, append-only table that represents the data flowing through Kafka.
  3.  
  4. CREATE STREAM page_views (
  5.     event_timestamp TIMESTAMP_LTZ(6),
  6.     user_id VARCHAR,
  7.     page VARCHAR
  8. ) WITH (
  9.     'topic'='csa_pageviews',
  10.     'value.format'='json',
  11.     'timestamp'='event_timestamp'
  12. );
  13.  
  14. CREATE STREAM cart_updates (
  15.     event_timestamp TIMESTAMP_LTZ(6),
  16.     user_id VARCHAR,
  17.     cart_action VARCHAR,
  18.     item_id VARCHAR
  19. ) WITH (
  20.     'topic'='csa_cart_updates',
  21.     'value.format'='json',
  22.     'timestamp'='event_timestamp'
  23. );
  24.  
  25. CREATE STREAM chat_messages (
  26.     event_timestamp TIMESTAMP_LTZ(6),
  27.     user_id VARCHAR,
  28.     message VARCHAR
  29. ) WITH (
  30.     'topic'='csa_chat_messages',
  31.     'value.format'='json',
  32.     'timestamp'='event_timestamp'
  33. );

Step 2: Build the Context with a Materialized View This is where the magic happens. We'll create a single MATERIALIZED VIEW that joins the three streams together and calculates key metrics in real-time. This view will be the single source of truth for our agent's context.

  1. -- Step 2: Create the Materialized View for the live customer context.
  2. -- This MV continuously processes events from the streams to build a real-time profile of each user.
  3. -- We first join all three streams and then calculate metrics over a 10-minute sliding window.
  4.  
  5. CREATE STREAM pv_cu
  6. WITH
  7.   ('kafka.topic.retention.ms' = '9172800000') AS
  8. SELECT
  9.   pv.event_timestamp AS event_timestamp,
  10.   pv.user_id AS user_id,
  11.   page,
  12.   cart_action AS cart_action,
  13.   item_id
  14. FROM
  15.   page_views AS pv
  16. WITH
  17.   ('timestamp' = 'event_timestamp')
  18.   JOIN cart_updates AS cu
  19. WITH
  20.   ('timestamp' = 'event_timestamp') WITHIN 10 SECONDS ON pv.user_id = cu.user_id;
  21.  
  22.  
  23. CREATE STREAM pv_cu_cm
  24. WITH
  25.   ('kafka.topic.retention.ms' = '9172800000') ASSELECT
  26.   pv_cu.event_timestamp AS event_timestamp,
  27.   pv_cu.user_id AS user_id,
  28.   page,
  29.   cart_action AS cart_action,
  30.   item_id,
  31.   cm.message
  32. FROM
  33.   pv_cu AS pv_cu
  34.   LEFT JOIN chat_messages AS cm WITHIN 10 SECONDS ON pv_cu.user_id = cm.user_id;
  35.  
  36.  
  37. CREATE MATERIALIZED VIEW live_customer_context_view AS
  38. SELECT
  39.   window_start,
  40.   window_end,
  41.   user_id,
  42.   -- Get the most recent page visited
  43.   LISTAGG (page) AS page_visited_list,
  44.   -- Count distinct pages visited in the last 10 minutes
  45.   COUNT(DISTINCT page) AS distinct_pages_visited_10m,
  46.   -- Count how many items were added to the cart in the last 10 minutes
  47.   SUM(
  48.     CASE
  49.       WHEN cart_action = 'ADD' THEN 1
  50.       ELSE 0
  51.     END
  52.   ) AS items_added_to_cart_10m,
  53.   -- Get the last chat message sent by the user
  54.   LISTAGG (message) AS chat_message_list
  55. FROM
  56.   HOP (pv_cu_cm, SIZE 10 MINUTE, ADVANCE BY 5 SECOND)
  57. WITH
  58.   (
  59.     'timestamp' = 'event_timestamp',
  60.     'starting.position' = 'earliest'
  61.   )
  62. GROUP BY
  63.   window_start,
  64.   window_end,
  65.   user_id;

You can now SELECT * FROM live_customer_context_view LIMIT 5; in DeltaStream and watch as your users' profiles update with every new event from Kafka. This live, structured context is the fuel our agent needs.

Part 3: The Context Hotline - Building the MCP Server

Our agent needs a simple way to ask for this context. We'll build a lightweight Java web server that provides a clean REST API endpoint. This server implements our "Model Context Protocol"—a consistent format for requesting and receiving context.

This server creates an endpoint GET /context/:userId. When called, it queries the materialized view in DeltaStream and returns the user's live profile as a JSON object. Remember to fill in your DeltaStream connection credentials correctly. Again, for brevity, we skip presenting the code here, but you can get the details in DeltaStream’s public examples github repo(https://github.com/deltastreaminc/examples/tree/main/ProactiveCustomerSupportAgent).

Part 4: The Final Piece - The Agent Takes Action

We now have all the plumbing in place. The final step is for the agentic AI application (which you might build using a framework like LangChain) to use this context.

While we won't build a full agent here, its core logic would look something like this in pseudocode:

  1. // Conceptual Agent Logic
  2. public class ProactiveAgent {
  3.  
  4.     public void decideNextAction(String userId) {
  5.         // 1. Call the MCP Server to get real-time context
  6.         Context currentUserContext = callMcpServer("/context/" + userId);
  7.  
  8.         // 2. Use the context to reason about the user's intent
  9.         boolean isStuck = currentUserContext.getDistinctPagesVisited10m() > 5 &&
  10.                           currentUserContext.getItemsAddedToCart10m() == 0;
  11.  
  12.         boolean isHesitatingAtCheckout = currentUserContext.getLastPageVisited().equals("/checkout") &&
  13.                                          currentUserContext.getLastChatMessage() == null;
  14.  
  15.         // 3. Take proactive action
  16.         if (isStuck) {
  17.             // Proactively open a chat window with a helpful message
  18.             sendProactiveChatMessage(userId, "Having trouble finding what you're looking for? I can help!");
  19.         } else if (isHesitatingAtCheckout) {
  20.             // Proactively offer a small discount to encourage purchase
  21.             offerDiscount(userId, "10_PERCENT_OFF");
  22.         }
  23.     }
  24. }

This simple logic shows the power of what we've built. The agent isn't waiting for a command. It's observing the user's real-time behavior, inferring their state of mind, and taking proactive steps to help.

Conclusion

We've successfully built a robust, end-to-end pipeline for a truly modern AI application. We moved beyond the simple, reactive chatbot and laid the foundation for a proactive AI agent.

By using Kafka for event transport, DeltaStream for powerful real-time processing, and a simple API server to serve context, we've enabled our AI to have the one thing it needs most to be truly helpful: awareness of the now. The next time you think about building an AI, don't just give it a brain—give it senses.

This blog was written by the author with assistance from AI to help with outlining, drafting, or editing.