Query a Lightning Table with MCP in Confluent Cloud

The Real-Time Context Engine is the Model Context Protocol (MCP) server for Lightning Tables. After you connect an AI agent to the Real-Time Context Engine, the agent uses MCP tools to discover topics, inspect schemas, and query data.

To enable a Lightning Table for a topic, see Enable a Lightning Table for a topic.

An AI agent should follow this workflow when querying the Real-Time Context Engine:

Call the listTopics tool to discover available topics.
Call the getMetadata tool for each relevant topic to learn column names and types.
Call the queryData tool to run queries against the topic.

The agent should call the getMetadata tool before the queryData tool to prevent errors from incorrect column names or types.

Connect an AI agent

After you enable a Lightning Table for a topic, you can connect an AI agent to query the data.

Get the MCP endpoint URL

The MCP endpoint URL for your cluster follows this format:

https://mcp.<region>.aws.confluent.cloud/mcp/v1/context-engine/organizations/<org_id>/environments/<env_id>/kafka-clusters/<lkc_id>

Replace the following placeholders:

<region>: The AWS region of your cluster, for example us-west-2.
<org_id>: Your Confluent Cloud organization ID.
<env_id>: Your Confluent Cloud environment ID.
<lkc_id>: Your Apache Kafka® cluster ID.

Note

If you connect to the Real-Time Context Engine over a private network, the endpoint format differs from the public endpoint format. For endpoint formats by networking option, see Private Networking.

Get a Base64 token

The Real-Time Context Engine authenticates MCP requests with HTTP Basic authentication, which requires a Base64 token. To create a Global API key, see Create a Global API key. To generate a token from it, see Generate the Base64 token.

The Real-Time Context Engine also supports signing in with your Confluent Cloud account and OAuth. For details, see Manage Access to Lightning Tables in Confluent Cloud.

Configure an MCP client

Use the MCP endpoint URL and your Base64 token to configure your MCP client.

Configure Claude Code

To add the Real-Time Context Engine MCP server in Claude Code, run the following command:

claude mcp add --transport http \
  confluent-rtce \
  <mcp_endpoint_url> \
  --header "Authorization: Basic <token>"

Replace the following placeholders:

<mcp_endpoint_url>: The MCP endpoint URL for your cluster. For the URL format, see Get the MCP endpoint URL.
<token>: The Base64 token you generated in Generate the Base64 token.

Configure any MCP client

Any MCP client that supports streamable HTTP transport can connect to the Real-Time Context Engine. Configure your client like the following example:

{
  "mcpServers": {
    "confluent-rtce": {
      "url": "https://mcp.<region>.aws.confluent.cloud/mcp/v1/context-engine/organizations/<org_id>/environments/<env_id>/kafka-clusters/<lkc_id>",
      "headers": {
        "Authorization": "Basic <token>"
      }
    }
  }
}

Run your first query

After you connect your MCP client, your AI agent can use three MCP tools: listTopics, getMetadata, and queryData. You can interact with these tools through natural-language prompts.

Try the following prompts to explore your data:

“What topics are available?”
“Describe the schema for my topic.”
“Show me the 10 most recent records in my topic.”

The AI agent translates your prompts into the appropriate MCP tool calls and returns the results.

For the full set of supported query patterns, including filters, ranges, and predicates with compound logic, see SQL Reference for Lightning Table Queries in Confluent Cloud.

MCP tools

The Real-Time Context Engine exposes three MCP tools that an AI agent calls to discover and query topic data.

listTopics

Returns the list of topics that have a Lightning Table and are visible to the agent. The listTopics tool takes no arguments.

The response includes the available topics and their descriptions.

getMetadata

Returns the schema and column metadata for a topic that has a Lightning Table. The agent should call the getMetadata tool before the queryData tool to learn column names and types.

The agent passes the name of the topic to inspect.

The response includes column names, data types, and primary key information for the topic.

queryData

Runs a read-only SQL query against a topic that has a Lightning Table. The agent passes the name of the topic to query and an SQL fragment that specifies the SELECT, WHERE, ORDER BY, and LIMIT clauses. The agent must also set the maximum number of rows to return. For the maximum value, see Query limits.

The response includes the matching rows. If the query fails, the response includes an error code and message.

Example queries

The following examples show requests you can make to an AI agent that’s connected to the Real-Time Context Engine. The agent translates each request into a read-only SQL query and returns the results.

“Look up order ORD-4521.”
“Show me all orders that haven’t been canceled.”
“Which orders are pending or processing?”
“Find orders that haven’t shipped yet.”
“Find orders between $100 and $500.”
“Show me completed orders over $100, sorted by highest total.”
“What are the top 10 highest-value completed orders?”

The agent generates SQL to fulfill these requests. For example, asking for “completed orders over $100, sorted by highest total” produces a query like the following:

SELECT ORDER_ID, STATUS, TOTAL
FROM orders_topic
WHERE STATUS = 'completed' AND TOTAL > 100
ORDER BY TOTAL DESC
LIMIT 10

Query limits

The Real-Time Context Engine enforces the following limits on queries:

Limit	Value
Maximum rows returned	200
Query timeout (seconds)	20

The max_result_rows parameter is required on every queryData call. Setting it to 0 or omitting it returns an error. The SQL LIMIT clause can further reduce the result set but can’t exceed the max_result_rows cap.

Rate limits also apply. If the agent exceeds the rate limit, the Real-Time Context Engine returns a RATE_LIMIT_EXCEEDED error.