---
title: "Connect Lenses to ChatGPT: Manage Kafka clusters and data pipelines"
slug: connect-lenses-to-chatgpt-manage-kafka-clusters-and-data-pipelines
date: 2026-07-16
author: Roopendra Talekar
categories: ["AI & Agents"]
excerpt: "Learn how to connect Lenses to chatgpt using Truto. Step-by-step guide to tool calling, API quirks, and autonomous workflows."
canonical: https://truto.one/blog/connect-lenses-to-chatgpt-manage-kafka-clusters-and-data-pipelines/
---

# Connect Lenses to ChatGPT: Manage Kafka clusters and data pipelines


If you need to connect Lenses to ChatGPT to automate Kafka cluster administration, manage data pipelines, or provision Access Control Lists (ACLs), you need a [Model Context Protocol (MCP) server](https://truto.one/what-is-model-context-protocol-mcp/). This server acts as a translation layer, interpreting natural language requests from ChatGPT and executing them as REST API calls against your Lenses environment. If your team uses Claude, check out our guide on [connecting Lenses to Claude](https://truto.one/connect-lenses-to-claude-audit-logs-acls-and-schema-governance/) or explore our broader architectural overview on [connecting Lenses to AI Agents](https://truto.one/connect-lenses-to-ai-agents-automate-kafka-connect-and-sql-streams/).

Giving a Large Language Model (LLM) read and write access to an enterprise Kafka environment is an engineering challenge with severe consequences for failure. You cannot afford an AI agent hallucinating a consumer group offset reset or deploying an untested SQL processor directly into production. You have to map highly nested JSON schemas to MCP tool definitions, manage strict API authorization, and deal with Kafka-specific state machine constraints. 

You can either spend weeks building and maintaining this custom middleware, or you can use Truto to dynamically generate a [managed MCP server](https://truto.one/mcp-as-a-service-managed-infrastructure-for-ai-agents/) URL based on Lenses's actual API documentation. This guide breaks down exactly how to use Truto to generate a managed MCP server for Lenses, connect it to ChatGPT, and execute complex data streaming workflows using natural language.

## The Engineering Reality of the Lenses API

A custom MCP server is a self-hosted integration layer. While the open MCP standard provides a predictable way for models to discover tools, the reality of implementing it against the Lenses API is difficult. You are not just building standard CRUD endpoints - you are interfacing with Kafka topics, [Kafka Connect](https://truto.one/how-to-manage-kafka-connect-pipelines-with-ai-agents/), Schema Registries, and SQL processors.

If you decide to build a custom MCP server for Lenses, you own the entire API lifecycle. Here are the specific integration challenges that break standard assumptions when working with Lenses:

### The Kafka State Machine Trap
Lenses enforces strict state requirements for mutating infrastructure. For example, if an AI agent wants to alter a consumer group offset, the consumer group must be in a `STOPPED` state. If your MCP server blindly passes the offset update payload to Lenses while the consumer is active, Lenses will reject the request. The LLM must be equipped with tools to query the operational state of a task or consumer group, stop it, execute the mutation, and restart it - all in the correct order.

### Composite Resource Routing
Almost every endpoint in the Lenses API requires strict composite routing. You cannot simply query `/api/connectors/123`. A Kafka Connect connector exists within a specific `environment` and a specific `cluster`. The AI agent must pass the environment name, the cluster name, and the connector ID for every operation. If your MCP server's schemas do not explicitly mark these composite keys as required parameters, the LLM will omit them, resulting in failed API calls and hallucinated recovery attempts.

### Strict Rate Limits and 429 Errors
When an AI agent is performing a cluster audit or summarizing topic metrics across a large environment, it can easily hit Lenses API rate limits. Truto does not retry, throttle, or apply backoff on rate limit errors. When the upstream Lenses API returns an HTTP 429, Truto passes that error directly to the caller. Truto normalizes the upstream rate limit information into [standardized headers](https://truto.one/handling-api-rate-limits-in-mcp-servers/) (`ratelimit-limit`, `ratelimit-remaining`, `ratelimit-reset`) per the IETF specification. The caller (the AI agent or MCP client framework) is entirely responsible for reading these headers and executing its own retry and exponential backoff logic.

## Generating a Managed Lenses MCP Server

Instead of building a custom server from scratch, you can use Truto to generate an MCP server dynamically. Truto reads the Lenses API documentation and resource schemas, parses them into JSON Schema format, and serves them over a JSON-RPC 2.0 endpoint.

You can create this MCP server in two ways.

### Method 1: Via the Truto UI

The fastest way to generate an MCP server is through the Truto dashboard.

1. Log into your Truto account and navigate to the integrated account page for your Lenses connection.
2. Click the **MCP Servers** tab.
3. Click **Create MCP Server**.
4. Configure the server. You can restrict the AI to read-only operations or filter by specific tags (like limiting it to `topics` and `alerts`).
5. Click Save and copy the generated MCP server URL (e.g., `https://api.truto.one/mcp/a1b2c3d4e5f6...`).

### Method 2: Via the Truto API

If you are provisioning AI infrastructure programmatically, you can generate the MCP server via a REST call to Truto. The API returns a secure, hashed token URL backed by Cloudflare KV.

```bash
curl -X POST https://api.truto.one/integrated-account/{integrated_account_id}/mcp \
  -H "Authorization: Bearer YOUR_TRUTO_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Lenses Cluster Admin Agent",
    "config": {
      "methods": ["read", "write", "custom"]
    }
  }'
```

The response will contain the `url` required for ChatGPT:

```json
{
  "id": "abc-123",
  "name": "Lenses Cluster Admin Agent",
  "config": { "methods": ["read", "write", "custom"] },
  "url": "https://api.truto.one/mcp/a1b2c3d4e5f6..."
}
```

## Connecting the MCP Server to ChatGPT

Once you have the Truto MCP URL, you must register it with your LLM framework.

### Method A: Via the ChatGPT UI

If you are using ChatGPT Pro, Plus, Enterprise, or Education accounts, you can add the server directly via Developer Mode.

1. In ChatGPT, navigate to **Settings** -> **Apps** -> **Advanced settings**.
2. Enable **Developer mode**.
3. Under MCP servers / Custom connectors, click to add a new server.
4. Enter a name (e.g., "Lenses Cluster Admin").
5. Paste the Truto MCP server URL into the Server URL field.
6. Click **Add**. ChatGPT will automatically complete the MCP handshake, call `tools/list`, and register the Lenses API endpoints.

### Method B: Via Manual Config File (Custom Agents / Desktop)

If you are building custom AI agents with LangChain, LlamaIndex, or using desktop environments that accept standard MCP JSON configurations, you can connect the server using a Server-Sent Events (SSE) transport.

```json
{
  "mcpServers": {
    "lenses-admin": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-sse",
        "--url",
        "https://api.truto.one/mcp/a1b2c3d4e5f6..."
      ]
    }
  }
}
```

## Hero Tools for Lenses

Truto automatically generates descriptive snake_case tool names based on the Lenses documentation. Here are the highest-leverage tools available to your [AI agent](https://truto.one/best-practices-for-designing-ai-agent-tools/).

### 1. list_all_lenses_proxy_topics
Retrieves a summary of all Kafka topics in a given environment, including partition counts, replication factors, and key/value schema types.

> "List all topics in the 'production-us-east' environment. Find any topics that have a replication factor of 1 and flag them as a high availability risk."

### 2. create_a_lenses_proxy_stream
Deploys a new SQL processor (stream) into the Lenses environment. The AI agent passes the exact SQL syntax, deployment namespace, and pipeline configurations.

> "Deploy a new SQL processor in 'production' named 'user-click-aggregation'. The SQL should select user_id and count(clicks) from the raw_clicks topic grouped by a 1-minute tumbling window, outputting to the 'aggregated_clicks' topic."

### 3. get_single_lenses_cluster_connector_by_id
Retrieves the detailed configuration, status, and running tasks for a specific Kafka Connect connector. This is vital for diagnosing failed data pipelines.

> "Check the status of the 'postgres-cdc-orders' connector in the 'primary' cluster. Are any of its tasks in a FAILED state?"

### 4. lenses_proxy_acls_bulk_update
Creates or updates a [Kafka Access Control List](https://truto.one/automate-kafka-acls-with-ai-agents/). The agent can use this to securely provision read or write access for specific principal users to specific topics.

> "Create an ACL in the 'staging' environment granting the principal 'User:data-eng-app' read access to all topics starting with 'analytics_events_'."

### 5. update_a_lenses_consumer_partition_offset_by_id
Alters the consumer group offset for a specific topic-partition. This is a highly privileged tool used for skipping poison pill messages or replaying data.

> "The consumer group 'billing-processor' is stuck on partition 4 of the 'invoices' topic. Update the offset for that partition to move it forward by 1 so we can bypass the poison pill message. Ensure the group is stopped first."

### 6. list_all_lenses_alert_events
Fetches paginated alert events from the Lenses environment, allowing the agent to act as an automated Level 1 support responder for cluster health.

> "Fetch the latest alert events for the 'production' environment over the last 4 hours. Summarize any consumer lag alerts and identify which consumer groups are affected."

To see the full schema details, required parameters, and the complete list of available endpoints, view the [Lenses integration page](https://truto.one/integrations/detail/lenses).

## Workflows in Action

When you give ChatGPT access to Lenses via MCP, it can chain these tools together to execute complex engineering investigations and remediation tasks.

### Scenario 1: Triaging a Data Pipeline Failure

**User Prompt:**
> "Check if any Kafka Connect tasks are failing in the 'eu-central' environment. If you find one, fetch the connector's configuration and attempt to restart the failed task."

**Execution Steps:**
1. **`list_all_lenses_kafka_connect_connections`**: The agent lists all connectors in the environment to find their states.
2. **`get_single_lenses_cluster_connector_by_id`**: For a connector marked with a FAILED task, the agent fetches the specific task ID and configuration details to understand what caused the failure.
3. **`create_a_lenses_task_restart`**: The agent submits a restart command for the specific failed task ID on the specific cluster.

*Result:* The LLM identifies the failed `elasticsearch-sink` task, reviews its config to ensure no obvious schema mismatch exists, restarts it, and confirms to the user that the pipeline is back online.

```mermaid
sequenceDiagram
    participant User as User
    participant ChatGPT as ChatGPT
    participant TrutoMCP as Truto MCP
    participant Lenses as Lenses API
    
    User->>ChatGPT: Check failing tasks in eu-central and restart them
    ChatGPT->>TrutoMCP: Call list_all_lenses_kafka_connect_connections(env=eu-central)
    TrutoMCP->>Lenses: GET /api/v1/connect/connections
    Lenses-->>TrutoMCP: Returns connector list (1 failed)
    TrutoMCP-->>ChatGPT: Connector data
    ChatGPT->>TrutoMCP: Call get_single_lenses_cluster_connector_by_id(env=eu-central, id=elastic-sink)
    TrutoMCP->>Lenses: GET /api/v1/connect/connections/elastic-sink
    Lenses-->>TrutoMCP: Returns task statuses
    TrutoMCP-->>ChatGPT: Task 2 is FAILED
    ChatGPT->>TrutoMCP: Call create_a_lenses_task_restart(env=eu-central, task_id=2)
    TrutoMCP->>Lenses: POST /api/v1/connect/connections/elastic-sink/tasks/2/restart
    Lenses-->>TrutoMCP: 204 No Content
    TrutoMCP-->>ChatGPT: Success
    ChatGPT-->>User: Identified task 2 on elastic-sink as failed. Restart triggered successfully.
```

### Scenario 2: Automated Microservice Provisioning

**User Prompt:**
> "We are launching a new microservice called 'inventory-service' in the 'staging' environment. Create a new topic called 'inventory-updates' with 3 partitions. Then, create an ACL giving the user 'inventory-service-app' Write access to this topic."

**Execution Steps:**
1. **`create_a_lenses_proxy_topic`**: The agent formats a JSON payload specifying `topicName: inventory-updates`, `partitions: 3`, and standard configs, executing it against the staging environment.
2. **`lenses_proxy_acls_bulk_update`**: The agent builds the ACL payload for `resourceType: Topic`, `resourceName: inventory-updates`, `principal: User:inventory-service-app`, `operation: Write`, and `permissionType: Allow`.

*Result:* In less than 10 seconds, the agent provisions the required Kafka infrastructure and applies the exact security governance rules required, completely removing the need for a developer to click through the Lenses UI.

```mermaid
flowchart TD
    A["User Prompt<br>Provision inventory-updates topic and ACLs"] --> B["ChatGPT parsing requirements"]
    B --> C["Call create_a_lenses_proxy_topic"]
    C --> D["Topic created with 3 partitions"]
    D --> E["Call lenses_proxy_acls_bulk_update"]
    E --> F["ACL granted for User:inventory-service-app"]
    F --> G["Return success summary to user"]
```

## Security and Access Control

Providing an LLM with administrative access to a Kafka cluster requires strict governance. Truto MCP servers operate entirely statelessly and provide several layers of access control at the token level:

*   **Method Filtering:** You can restrict a specific MCP server to read-only operations. By setting `config.methods: ["read"]`, the LLM can only query topics and metrics, strictly preventing it from creating streams or altering offsets.
*   **Tag Filtering:** Lenses resources can be grouped by tags. You can configure an MCP server to only expose tools tagged with `alerts` or `schemas`, keeping sensitive cluster administration endpoints hidden from the LLM context.
*   **Mandatory API Authentication:** By enabling `require_api_token_auth: true` on the MCP token, possession of the URL is not enough to execute a tool. The client framework must pass a valid Truto API token in the Authorization header, preventing anonymous network execution.
*   **Automatic Expiration:** If you are giving a contractor or temporary AI agent access to cluster diagnostics, you can set an `expires_at` timestamp. Truto schedules a cleanup alarm and automatically destroys the Cloudflare KV token record at the exact specified time.

## Wrap Up

Administering Kafka clusters requires deep context, precise syntax, and rapid incident response. By connecting Lenses to ChatGPT via a Truto MCP server, you transform your LLM from a generic chatbot into a highly capable site reliability engineer.

Instead of wasting engineering cycles managing OAuth tokens, complex pagination states, and 429 backoff logic, Truto handles the [protocol translation](https://truto.one/bridge-apis-to-llms-with-mcp-translation-layers/) so your team can focus on actual data infrastructure.

:::cta{buttonText="Talk to us" buttonUrl="https://cal.com/truto/partner-with-truto"} 
Want to automate your Lenses cluster administration with AI? Talk to our engineering team today.
:::