Streaming Agents Examples with Confluent Cloud

Confluent Intelligence supports Streaming Agents implementations that range from basic agent creation to complex multi-agent workflows. You can implement Streaming Agents with Flink SQL statements.

• Quick Start

• Customer Support Agent

• Financial Fraud Detection

• IoT Predictive Maintenance

• Content Moderation System

• Multi-Agent Chat System

• Reflection Agent

• Agent Chaining

• Conditional Agent Selection

• Error Handling and Fallbacks

• Best Practices

Quick start repo

Streaming Agents quickstart architecture

Use the Streaming Agents Quickstart repo to build your first Streaming Agent in minutes.

Customer support agent

Build a customer support agent that can answer questions and create support tickets.

The following diagram shows the high-level data flow for this example. Customer messages in an input topic trigger the agent, which uses a large language model (LLM) and tools to produce a response on an output topic.

customer-messages topic  -->  CREATE AGENT customer_support_agent  -->  support-responses topic
(input)                       (LLM + tools: customer_lookup,            (output)
                               support_tools)

Prerequisites

• Confluent Cloud account. If you don’t already have one, see Confluent Cloud Free Trial.

• Flink compute pool. To create one, see Create a compute pool.

• OpenAI API key

Step 1: Set up the environment

In this step, you create connections to the OpenAI API and the MCP endpoint.

-- Create connection to OpenAI
CREATE CONNECTION openai_connection
WITH (
  'type' = 'openai',
  'endpoint' = 'https://api.openai.com/v1/chat/completions',
  'api-key' = 'your-openai-key'
);

-- Create connection to support API
CREATE CONNECTION support_api
WITH (
  'type' = 'mcp_server',
  'endpoint' = 'https://support-api.example.com',
  'api-key' = '<your-support-api-key>'
);

Step 2: Create Tools

In this step, you create tools for the agent. This example creates a function-based tool for account lookup and an MCP-based tool for support API.

The customer-lookup.jar file isn’t provided; it must contain your own implementation of the account lookup logic, and you must upload it to Confluent Cloud before running the following statement. For more information, see Create a User-Defined Function.

-- Create function for account lookup. The JAR must already be uploaded.
CREATE FUNCTION lookup_customer
USING JAR 'customer-lookup.jar'
COMMENT 'Lookup customer information by ID';

-- Create function tool
CREATE TOOL customer_lookup
USING FUNCTION lookup_customer
WITH (
  'type' = 'function',
  'description' = 'Lookup customer account information'
);

-- Create MCP tool for support API
CREATE TOOL support_tools
USING CONNECTION support_api
WITH (
  'type' = 'mcp',
  'allowed_tools' = 'create_ticket,update_ticket,search_kb',
  'request_timeout' = '30'
);

To confirm that both tools were created, run SHOW TOOLS. The customer_lookup and support_tools tools should appear in the output.

Step 3: Create the model

In this step, you create a model that uses the OpenAI API and an agent that uses the model and the tools you created in the previous step.

CREATE MODEL support_llm
INPUT (message STRING)
OUTPUT (response STRING)
WITH (
  'provider' = 'openai',
  'task' = 'text_generation',
  'openai.connection' = 'openai_connection',
  'openai.model_version' = 'gpt-4o',
  'openai.temperature' = '0.7'
);

To confirm that the model was created, run SHOW MODELS. The support_llm model should appear in the output.

Step 4: Create the agent

In this step, you create an agent that uses the model and the tools you created in the previous step.

-- 'max_iterations' determines how many times the agent can call a tool while
-- handling a single message, to prevent runaway tool-calling loops.
CREATE AGENT customer_support_agent
USING MODEL support_llm
USING PROMPT 'You are a helpful customer support agent. Use the available tools to look up customer information and create support tickets when needed. Always be polite and professional.'
USING TOOLS customer_lookup, support_tools
WITH (
  'max_iterations' = '5'
);

To confirm that the agent was created, run SHOW AGENTS. The customer_support_agent agent should appear in the output.

Step 5: Create input and output tables

In this step, you create input and output tables for the agent.

-- Input table for customer messages
CREATE TABLE customer_messages (
  message_id STRING,
  customer_id STRING,
  message STRING,
  timestamp TIMESTAMP(3),
  WATERMARK FOR timestamp AS timestamp - INTERVAL '5' SECOND
) WITH (
  'topic' = 'customer-messages',
  'format' = 'json',
  'scan.startup.mode' = 'earliest'
);

-- Output table for agent responses
CREATE TABLE support_responses (
  message_id STRING,
  customer_id STRING,
  original_message STRING,
  agent_response STRING,
  ticket_id STRING,
  timestamp TIMESTAMP(3)
) WITH (
  'topic' = 'support-responses',
  'format' = 'json'
);

A message that’s produced to the customer-messages topic looks like the following:

{
  "message_id": "msg-1001",
  "customer_id": "cust-42",
  "message": "I was charged twice for my last order. Can you help?",
  "timestamp": "2026-09-15T18:32:00Z"
}

The agent’s raw response is a JSON-encoded string, so agent_response stores that string as-is, and ticket_id is a value extracted from it with JSON_EXTRACT. After the agent processes the message, the corresponding row in support-responses looks like the following:

{
  "message_id": "msg-1001",
  "customer_id": "cust-42",
  "original_message": "I was charged twice for my last order. Can you help?",
  "agent_response": "{\"reply\": \"I've looked into your account and confirmed the duplicate charge. I've opened a support ticket to process your refund.\", \"ticket_id\": \"tick-5521\"}",
  "ticket_id": "tick-5521",
  "timestamp": "2026-09-15T18:32:04Z"
}

Step 6: Run the agent

In this step, you run the agent on the input table and insert the results into the output table.

-- LATERAL TABLE(AI_RUN_AGENT(...)) invokes 'customer_support_agent' once
-- for every row read from customer_messages, passing t.message as the
-- agent's input and t.message_id to correlate the call. The function
-- returns a status and a response column for each invocation.
INSERT INTO support_responses
SELECT
  t.message_id,
  t.customer_id,
  t.message AS original_message,
  a.response AS agent_response,
  JSON_EXTRACT(a.response, '$.ticket_id') AS ticket_id,
  t.timestamp
FROM
  customer_messages AS t,
  LATERAL TABLE(
    AI_RUN_AGENT('customer_support_agent', t.message, t.message_id)
  ) AS a(status, response);

Financial fraud detection

Build a multi-agent system for real-time fraud detection.

Step 1: Create specialized agents

In this step, you create specialized agents for the fraud detection system.

-- Risk assessment agent
CREATE AGENT risk_assessor
USING MODEL fraud_llm
USING PROMPT 'You are a financial risk assessment expert. Analyze transactions for potential fraud indicators.'
USING TOOLS transaction_analyzer, risk_calculator
WITH (
  'max_iterations' = '3'
);

-- Investigation agent
CREATE AGENT fraud_investigator
USING MODEL fraud_llm
USING PROMPT 'You are a fraud investigation specialist. Investigate suspicious transactions and gather evidence.'
USING TOOLS customer_history, external_verification, alert_system
WITH (
  'max_iterations' = '8'
);

-- Decision agent
CREATE AGENT fraud_decision
USING MODEL fraud_llm
USING PROMPT 'You are a fraud decision maker. Based on investigation results, decide whether to block, flag, or approve transactions.'
USING TOOLS transaction_blocker, notification_system
WITH (
  'max_iterations' = '5'
);

Step 2: Create the workflow

In this step, you create a workflow that uses the agents you created in the previous step.

-- First stage: Risk assessment
CREATE VIEW risk_assessed_transactions AS
SELECT
  t.transaction_id,
  t.customer_id,
  t.amount,
  t.transaction_data,
  t.risk_score,
  a.response AS risk_analysis
FROM
  transactions AS t,
  LATERAL TABLE(
    AI_RUN_AGENT('risk_assessor', t.transaction_data, t.transaction_id)
  ) AS a(status, response);

-- Second stage: Investigation for high-risk transactions
CREATE VIEW investigated_transactions AS
SELECT
  t.transaction_id,
  t.customer_id,
  t.amount,
  t.transaction_data,
  t.risk_score,
  t.risk_analysis,
  a.response AS investigation_results
FROM
  risk_assessed_transactions AS t,
  LATERAL TABLE(
    AI_RUN_AGENT('fraud_investigator', t.transaction_data, t.transaction_id)
  ) AS a(status, response)
WHERE
  t.risk_score > 0.7;

-- Third stage: Final decision
INSERT INTO fraud_decisions
SELECT
  t.transaction_id,
  t.customer_id,
  t.amount,
  t.risk_score,
  a.response AS final_decision,
  JSON_EXTRACT(a.response, '$.action') AS action,
  JSON_EXTRACT(a.response, '$.confidence') AS confidence
FROM
  investigated_transactions AS t,
  LATERAL TABLE(
    AI_RUN_AGENT('fraud_decision', t.transaction_data, t.transaction_id)
  ) AS a(status, response);

IoT predictive maintenance

Build an agent system for predictive maintenance of IoT devices.

Step 1: Create data processing pipeline

-- Process sensor data
CREATE VIEW processed_sensor_data AS
SELECT
  device_id,
  sensor_type,
  value,
  timestamp,
  LAG(value) OVER (PARTITION BY device_id, sensor_type ORDER BY timestamp) as prev_value,
  value - LAG(value) OVER (PARTITION BY device_id, sensor_type ORDER BY timestamp) as value_change
FROM sensor_readings;

-- Aggregate data for analysis
CREATE VIEW device_health_summary AS
SELECT
  device_id,
  COUNT(*) as reading_count,
  AVG(value) as avg_value,
  STDDEV(value) as value_volatility,
  MAX(value) as max_value,
  MIN(value) as min_value,
  MAX(timestamp) as last_reading
FROM processed_sensor_data
WHERE timestamp >= CURRENT_TIMESTAMP - INTERVAL '1' HOUR
GROUP BY device_id;

Step 2: Create maintenance agent

In this step, you create a maintenance agent that uses the model and the tools you created in the previous step.

CREATE AGENT maintenance_agent
USING MODEL iot_llm
USING PROMPT 'You are an IoT maintenance expert. Analyze device health data and predict maintenance needs. Use the available tools to check device specifications and maintenance history.'
USING TOOLS device_specs, maintenance_history, parts_inventory
WITH (
  'max_iterations' = '6'
);

Step 3: Execute maintenance analysis

In this step, you run the maintenance agent on the device health summary table and insert the results into the maintenance_recommendations table.

INSERT INTO maintenance_recommendations
SELECT
  t.device_id,
  t.reading_count,
  t.avg_value,
  t.value_volatility,
  a.response AS maintenance_analysis,
  JSON_EXTRACT(a.response, '$.recommendation') AS recommendation,
  JSON_EXTRACT(a.response, '$.urgency') AS urgency,
  JSON_EXTRACT(a.response, '$.parts_needed') AS parts_needed
FROM
  device_health_summary AS t,
  LATERAL TABLE(
    AI_RUN_AGENT(
      'maintenance_agent',
      CONCAT(
        'device_id=', t.device_id,
        '; reading_count=', CAST(t.reading_count AS STRING),
        '; avg_value=', CAST(t.avg_value AS STRING),
        '; value_volatility=', CAST(t.value_volatility AS STRING),
        '; max_value=', CAST(t.max_value AS STRING),
        '; min_value=', CAST(t.min_value AS STRING)
      ),
      t.device_id
    )
  ) AS a(status, response);

Content moderation system

Build a content moderation system with multiple specialized agents.

Step 1: Create content analysis agents

In this step, you create content analysis agents that use the model and the tools you created in the previous step.

-- Text content analyzer
CREATE AGENT text_moderator
USING MODEL content_llm
USING PROMPT 'You are a content moderation expert specializing in text analysis. Identify inappropriate content, hate speech, and policy violations.'
USING TOOLS text_analyzer, policy_checker
WITH (
  'max_iterations' = '4'
);

-- Image content analyzer
CREATE AGENT image_moderator
USING MODEL vision_llm
USING PROMPT 'You are a content moderation expert specializing in image analysis. Identify inappropriate images, violence, and policy violations.'
USING TOOLS image_analyzer, content_classifier
WITH (
  'max_iterations' = '4'
);

-- Decision maker
CREATE AGENT moderation_decision
USING MODEL content_llm
USING PROMPT 'You are a content moderation decision maker. Based on analysis results, decide whether to approve, flag, or reject content.'
USING TOOLS content_action, user_notification
WITH (
  'max_iterations' = '3'
);

Step 2: Create Content Processing Pipeline

In this step, you create content processing pipeline that uses the agents you created in the previous step.

-- Process text content
CREATE VIEW text_analysis AS
SELECT
  t.content_id,
  t.user_id,
  t.content_type,
  t.text_content,
  a.response AS text_analysis_result
FROM
  user_content AS t,
  LATERAL TABLE(
     AI_RUN_AGENT('text_moderator', t.text_content, t.content_id)
  ) AS a(status, response)
WHERE
  t.content_type = 'text';

-- Process image content
CREATE VIEW image_analysis AS
SELECT
  t.content_id,
  t.user_id,
  t.content_type,
  t.image_url,
  a.response AS image_analysis_result
FROM
  user_content AS t,
  LATERAL TABLE(
    AI_RUN_AGENT('image_moderator', t.image_url, t.content_id)
  ) AS a(status, response)
 WHERE
   t.content_type = 'image';

-- Combine analysis results
CREATE VIEW combined_analysis AS
SELECT
  content_id,
  user_id,
  content_type,
  COALESCE(text_analysis_result, image_analysis_result) as analysis_result
FROM text_analysis
FULL OUTER JOIN image_analysis USING (content_id);

-- Make final decisions
INSERT INTO moderation_decisions
SELECT
  t.content_id,
  t.user_id,
  t.content_type,
  a.response AS final_decision,
  JSON_EXTRACT(a.response, '$.action') AS action,
  JSON_EXTRACT(a.response, '$.reason') AS reason,
  JSON_EXTRACT(a.response, '$.confidence') AS confidence
FROM
  combined_analysis AS t,
  LATERAL TABLE(
    AI_RUN_AGENT('moderation_decision', t.analysis_result, t.content_id)
  ) AS a(status, response);

Multi-agent chat system

Build a conversational AI system with multiple specialized agents.

Step 1: Create specialized agents

In this step, you create specialized agents that use the model and the tools you created in the previous step.

-- General conversation agent
CREATE AGENT conversation_agent
USING MODEL chat_llm
USING PROMPT 'You are a helpful conversational AI. Engage in natural conversation and route complex queries to appropriate specialists.'
USING TOOLS query_classifier, agent_router
WITH (
  'max_iterations' = '3'
);

-- Technical support agent
CREATE AGENT tech_support
USING MODEL tech_llm
USING PROMPT 'You are a technical support specialist. Help users with technical issues and provide detailed solutions.'
USING TOOLS knowledge_base, system_diagnostics, ticket_creator
WITH (
  'max_iterations' = '6'
);

-- Sales agent
CREATE AGENT sales_agent
USING MODEL sales_llm
USING PROMPT 'You are a sales specialist. Help customers find the right products and services for their needs.'
USING TOOLS product_catalog, pricing_calculator, quote_generator
WITH (
  'max_iterations' = '5'
);

Step 2: Create conversation management

In this step, you create conversation management table that uses the agents you created in the previous step.

-- Conversation state table
CREATE TABLE conversation_state (
  session_id STRING,
  user_id STRING,
  current_agent STRING,
  conversation_history STRING,
  last_activity TIMESTAMP(3),
  PRIMARY KEY (session_id) NOT ENFORCED
) WITH (
  'topic' = 'conversation-state',
  'changelog.mode' = 'upsert',
  'key.format' = 'json',
  'value.format' = 'json'
);

-- Process incoming messages
INSERT INTO conversation_state
SELECT
  t.session_id,
  t.user_id,
  CASE
    WHEN JSON_EXTRACT(a.response, '$.specialist') = 'tech_support' THEN 'tech_support'
    WHEN JSON_EXTRACT(a.response, '$.specialist') = 'sales' THEN 'sales_agent'
    ELSE 'conversation_agent'
  END AS current_agent,
  CONCAT(t.conversation_history, '\nUser: ', t.message, '\nAgent: ', a.response) AS conversation_history,
  CURRENT_TIMESTAMP AS last_activity
FROM
  user_messages AS t,
  LATERAL TABLE(
    AI_RUN_AGENT('conversation_agent', t.message, t.session_id)
  ) AS a(status, response);

Step 3: Route to specialized agents

In this step, you route to specialized agents that use the agents you created in the previous step.

-- Route to technical support
INSERT INTO tech_support_responses
SELECT
  session_id,
  user_id,
  message,
  a.response as response
FROM
user_messages AS t,
LATERAL TABLE(
    AI_RUN_AGENT('tech_support', t.message, t.session_id)
  ) AS a(status, response)
WHERE t.session_id IN (
  SELECT session_id
  FROM conversation_state
  WHERE current_agent = 'tech_support'
) AND a.status = 'SUCCESS';

-- Route to sales agent
INSERT INTO sales_responses
SELECT
  t.session_id,
  t.user_id,
  t.message,
  a.response AS response
FROM
  user_messages AS t,
  LATERAL TABLE(
    AI_RUN_AGENT('sales_agent', t.message, t.session_id)
  ) AS a(status, response)
WHERE
  t.session_id IN (
    SELECT session_id
    FROM conversation_state
    WHERE current_agent = 'sales_agent'
  );

Reflection agent

Build a reflection agent that generates a marketing campaign root-cause analysis and iteratively refines the output through a drafter-critic loop.

-- Create a model.
CREATE MODEL analysis_model
INPUT (message STRING)
OUTPUT (response STRING)
WITH (
  'provider' = 'openai',
  'task' = 'text_generation',
  'openai.connection' = 'openai_connection',
  'openai.model_version' = 'gpt-4o'
);

-- Drafter: performs root-cause analysis.
CREATE AGENT campaign_analyst
USING MODEL analysis_model
USING PROMPT 'You are a marketing analyst. Analyze the campaign
  performance data and identify root causes for underperformance.
  Provide a structured JSON response with fields: root_cause,
  contributing_factors, affected_segments, and
  recommended_actions.';

-- Critic: validates the analysis.
CREATE AGENT campaign_critic
USING MODEL analysis_model
USING PROMPT 'You are a senior marketing strategist. Evaluate
  the root-cause analysis for completeness. Verify that the
  analysis considers segmentation by device type, ad creative,
  and geographic region. Check that recommended actions are
  specific and actionable.';

-- Reflection agent.
CREATE AGENT campaign_rca
USING AGENTS campaign_analyst, campaign_critic
WITH (
  'type' = 'reflection',
  'pass_condition' = 'approval',
  'max_iterations' = '3'
);

-- Run the reflection agent.
INSERT INTO campaign_analysis_results
SELECT
  c.campaign_id,
  r.response AS analysis
FROM underperforming_campaigns AS c,
LATERAL TABLE(
  AI_RUN_AGENT(
    'campaign_rca',
    c.performance_data,
    CAST(c.campaign_id AS STRING)
  )
) AS r(status, response);

For more information about reflection workflows, refer to Improve Agent Output with Reflection Workflows.

Agent chaining

Chain multiple agents together for complex workflows:

-- First agent: Classify request
CREATE VIEW classified_requests AS
SELECT
  t.request_id,
  t.request_text,
  a.response AS classification
FROM
  user_requests AS t,
  LATERAL TABLE(
    AI_RUN_AGENT('request_classifier', t.request_text, t.request_id)
  ) AS a(status, response);

-- Second agent: Process based on classification
CREATE VIEW processed_requests AS
SELECT
  t.request_id,
  t.request_text,
  t.classification,
  a.response AS response
FROM
  classified_requests AS t,
  LATERAL TABLE(
    AI_RUN_AGENT('request_processor', t.request_text, t.request_id)
  ) AS a(status, response);

Conditional agent selection

The following example code selects different agents based on conditions.

CREATE VIEW dynamic_agent_responses AS
SELECT
  t.request_id,
  t.request_type,
  COALESCE(tech.response, sales.response, general.response) AS response
FROM
  user_requests AS t
LEFT JOIN LATERAL TABLE(
  AI_RUN_AGENT('tech_agent', t.request_text, t.request_id)
) AS tech(status, response)
  ON t.request_type = 'technical'
LEFT JOIN LATERAL TABLE(
  AI_RUN_AGENT('sales_agent', t.request_text, t.request_id)
) AS sales(status, response)
  ON t.request_type = 'sales'
LEFT JOIN LATERAL TABLE(
  AI_RUN_AGENT('general_agent', t.request_text, t.request_id)
) AS general(status, response)
  ON t.request_type NOT IN ('technical', 'sales');

Error handling and fallbacks

The following example code implements robust error handling.

CREATE AGENT robust_agent
USING MODEL main_llm
USING PROMPT 'You are a robust agent that handles errors gracefully. If a tool fails, try alternative approaches or ask for clarification.'
USING TOOLS primary_tool, fallback_tool, error_handler
WITH (
  'max_iterations' = '10',
  'request_timeout' = '600',
  'max_consecutive_failures' = '3',
  'handle_exception' = 'continue'
);

In this configuration, max_consecutive_failures sets how many times a tool, model, or remote user-defined function (UDF) call can fail consecutively before robust_agent stops trying the current message.

For example, if primary_tool fails three times in a row, the agent stops retrying that message.

handle_exception then determines what happens next. With 'handle_exception' = 'continue', the agent moves on and processes the next message instead of stopping the job. The failed message is marked with a failure status rather than a successful response. For more information about these options, see Error Handling.

Best practices

The following are best practices for designing and implementing agents.

Agent design

• Keep agents focused on specific tasks

• Use clear and descriptive prompts

• Limit the number of tools per agent

• Implement proper error handling

Performance optimization

• Use function-based tools for high-frequency operations

• Implement caching where appropriate

• Monitor execution times and costs

• Use appropriate parallelism

Security

• Store API keys securely

• Validate all inputs

• Implement proper access controls

• Monitor for security issues

Monitoring

• Set up comprehensive logging

• Monitor agent performance

• Track tool usage and costs

• Implement alerting for failures

Related content

• Streaming Agents Overview

• Build and Run Streaming Agents

• Improve Agent Output with Reflection Workflows

• Monitor Streaming Agents

• Debug Streaming Agents

• Agent Runtime Guide

• CREATE AGENT Statement

• CREATE TOOL Statement

• AI_RUN_AGENT Function

Note

This website includes content developed at the Apache Software Foundation under the terms of the Apache License v2.