Connect Lenses to AI Agents: Automate Kafka Connect & SQL Streams

You want to connect Lenses to an AI agent so your system can independently manage Kafka topics, restart failed Kafka Connect tasks, deploy SQL streams, and enforce DataOps governance. Here is exactly how to do it using Truto's /tools endpoint and SDK, bypassing the need to manually code dozens of endpoints or maintain complex Kafka metadata wrappers.

Giving a Large Language Model (LLM) read and write access to your Lenses instance is an engineering headache. You either spend weeks building, hosting, and maintaining a custom connector that understands the nuances of Kafka state management, or you use a managed infrastructure layer that handles the boilerplate for you. If your team uses ChatGPT, check out our guide on connecting Lenses to ChatGPT, or if you are building on Anthropic's models, read our guide on connecting Lenses to Claude. For developers building custom autonomous workflows, you need a programmatic way to fetch these tools and bind them to your agent framework.

This guide breaks down exactly how to fetch AI-ready tools for Lenses, bind them natively to an LLM using LangChain (or any framework like LangGraph, CrewAI, or the Model Context Protocol), and execute complex DataOps workflows. For a deeper look at the architecture behind this approach, refer to our research on architecting AI agents and the SaaS integration bottleneck.

The Engineering Reality of Custom Lenses Connectors

Building AI agents is easy. Connecting them to external DataOps APIs is hard. Giving an LLM access to external data sounds simple in a prototype. You write a Node.js function that makes a fetch request and wrap it in an @tool decorator to leverage LLM function calling. In production, this approach collapses entirely, especially with a platform as complex as Lenses.

If you decide to build the integration yourself, you own the entire API lifecycle. Lenses acts as a control plane for Apache Kafka, which introduces several highly specific integration challenges that break standard LLM assumptions.

Multi-Part Identifiers and Environment Awareness

Unlike a typical CRM where an ID is a simple GUID, Lenses operations frequently require a matrix of identifiers. To get a specific Kafka connection, the agent must pass the environment, the api_version, and the id. When an LLM hallucinations an API version or forgets to pass the environment string, the entire execution halts. Hardcoding this context into your agent prompts inflates token costs and reduces accuracy.

State Machine Constraints

Kafka operations are heavily state-dependent. If you instruct an LLM to alter the consumer offsets for a Kafka Connect connector, the upstream Kafka cluster requires that connector to be in a STOPPED state. An LLM directly hitting raw endpoints will blindly issue an offset update request, fail with a cryptic 400 error, and likely enter an infinite retry loop. The integration layer needs to present these operations as discrete, safe tools rather than raw endpoints.

Complex Nested Payloads for SQL Streams

Deploying a SQL processor (K2K app) via the Lenses API requires a deeply nested YAML or JSON manifest (apiVersion, kind, metadata, spec). If the LLM misses a required nested key like deployment_namespace_lrn, the upstream API rejects it. By exposing the Lenses API through a proxy layer that converts these endpoints into strict JSON Schema tools, you force the LLM to conform to the required structure before the network request is ever made.

Why a Unified Tool Layer Matters for Agent Safety

Direct API tools (one tool per raw Lenses endpoint) push provider quirks into the LLM's context. A unified tool layer collapses these complexities behind standardized schemas.

Your agent sees list_all_lenses_kafka_connect_connectors and create_a_lenses_task_restart - with perfectly structured required arguments. That gives you concrete safety wins:

1. Deterministic input validation. Every tool has a strict JSON schema. Invalid arguments are rejected locally before they hit the Lenses environment.
2. Clean error boundaries. When things fail, the agent gets a structured error it can parse, not a raw HTML error page or an obscure Java stack trace.
3. Simplified pagination and authentication. The proxy layer handles the boilerplate of iterating through pages of Kafka topic metrics or managing Lenses API tokens.

flowchart TD
    A["LLM Framework<br>(LangChain/LangGraph)"] -->|"Calls tool with JSON"| B["Truto Tool Layer<br>(Proxy API)"]
    B -->|"Injects Auth + Validates Schema"| C["Lenses API<br>(Upstream)"]
    C -->|"Raw Data"| B
    B -->|"Normalized JSON"| A

Fetching AI-Ready Tools for Lenses

Truto provides a dedicated /tools endpoint that exposes all defined methods on an integration as AI-ready tools. You don't have to map the Lenses Swagger file manually.

When you call GET https://api.truto.one/integrated-account/<id>/tools, Truto returns a list of tools formatted with names, descriptions, and strict JSON schemas for the parameters. You can filter this list using query parameters. For example, if you only want read-only methods for a diagnostic agent, you can append ?methods [0]=read.

The Hero Lenses Agent Tools

Exposing the entire Lenses API to a single agent usually overwhelms the context window. Instead, you should arm your agent with high-leverage "hero tools" that map to your core DataOps workflows. Here are the most powerful Lenses tools you can provide to an LLM.

1. List Kafka Connect Connectors

Tool Name: list_all_lenses_kafka_connect_connectors

This tool retrieves all Kafka Connect connectors across your environments. It is the foundational tool for any monitoring or auto-remediation agent. It returns an array of connector objects containing their current status, state, and task health.

"Get the current status of all Kafka Connect connectors in the production environment. Identify any connectors that are in a FAILED or DEGRADED state."

2. Restart a Connector Task

Tool Name: create_a_lenses_task_restart

Kafka Connect tasks routinely fail due to transient network issues or bad records. Instead of paging an on-call engineer, an agent can use this tool to automatically issue a restart command to the specific failed task within a connector cluster.

"Restart task ID 2 for the 'postgres-source-connector' in the primary environment."

3. Create a Kafka Topic

Tool Name: create_a_lenses_proxy_topic

Allow agents to handle self-service infrastructure requests. When a developer asks for a new topic, the agent can validate the request against naming conventions, determine the appropriate replication factor and partition count, and execute the creation.

"Create a new Kafka topic named 'user-signup-events-v1' with 6 partitions and standard retention configs in the staging environment."

4. Manage Access Control Lists (ACLs)

Tool Name: lenses_proxy_acls_bulk_update

Security and governance are critical in Kafka. This tool allows an agent to dynamically provision or update Kafka Access Control Lists (ACLs) for specific principals, ensuring least-privilege access to sensitive topics.

"Grant read-only ACL permissions to the principal 'User:analytics-service' for the 'financial-transactions' topic in the production environment."

5. Execute Proxy Requests (Direct SQL)

Tool Name: create_a_lenses_environment_proxy

This is a highly versatile tool. It proxies a raw request to a Lenses environment, allowing the agent to execute Lenses SQL directly against live data. This is invaluable for data profiling, message inspection, or verifying that a topic is actively receiving the correct schema.

"Run a SQL query against the Lenses proxy to select the last 5 messages from the 'order-events' topic where the total amount is greater than 1000."

6. Alter Connector Offsets

Tool Name: lenses_connector_offsets_bulk_update

When a sink connector gets stuck on a poison pill message, the standard operational procedure is to alter the offset to skip the bad record. This tool allows the agent to update the offsets programmatically. Note: The agent must ensure the connector is stopped before executing this tool.

"Update the consumer offset for the 'elasticsearch-sink-connector' to skip partition 3 forward by one record."

To view the exact JSON schemas, required properties, and the full list of available operations, visit the Lenses integration page.

Workflows in Action

Individual tools are useful, but the real power of AI agents emerges when tools are chained together to execute autonomous workflows. Here is how specific engineering personas use these Lenses tools in production.

Scenario 1: Automated Incident Remediation (DataOps Engineer)

An on-call engineer receives a PagerDuty alert that data has stopped flowing into Snowflake. Instead of manually logging into the Lenses UI, checking connectors, and restarting tasks, an autonomous agent handles the triage.

"Diagnose the 'snowflake-sink-connector'. Check its status and restart any failed tasks. If it is still failing, check the offset position and skip the poison pill record if necessary."

Tool Execution Sequence:

1. list_all_lenses_kafka_connect_connectors - The agent identifies that the 'snowflake-sink-connector' is degraded.
2. get_single_lenses_cluster_connector_by_id - The agent drills down into the specific connector and sees that Task 1 has failed.
3. create_a_lenses_task_restart - The agent issues a restart for Task 1.
4. list_all_lenses_kafka_connect_connectors - The agent loops back to verify the task has returned to a RUNNING state.

Result: The agent successfully restarts the stalled task, validates that the data pipeline is functioning again, and logs the remediation in the incident ticket without human intervention.

Scenario 2: Self-Service Provisioning (Platform Engineer)

A developer opens a Jira ticket requesting a new Kafka topic and corresponding read access for their microservice. The agent intercepts the ticket, validates the request against company policies, and executes the infrastructure changes.

"Review the new infrastructure request. Provision the requested Kafka topic 'customer-profile-updates', configure it for compaction, and grant the 'profile-service' read access."

Tool Execution Sequence:

1. create_a_lenses_proxy_topic - The agent creates the topic, explicitly passing the configuration flags for log compaction.
2. lenses_proxy_acls_bulk_update - The agent assigns an ACL allowing User:profile-service to execute READ operations on the new topic.
3. list_all_lenses_proxy_topics - The agent verifies the topic was created and is visible in the cluster.

Result: The developer's request is fulfilled in seconds. The agent replies to the ticket with the connection details, completely eliminating the Platform Engineering team's operational bottleneck.

Building Multi-Step Workflows

To build these multi-step workflows, you need to bind Truto's dynamically generated Lenses tools to your LLM. The exact implementation depends on your framework, but the architectural pattern remains the same: initialize the SDK, fetch the tools, bind them to the model, and execute the agent loop.

The following example uses LangChain.js and the truto-langchainjs-toolset SDK to create an agent that can interact with Lenses.

sequenceDiagram
    participant App as Agent App
    participant SDK as TrutoToolManager
    participant LLM as LLM (OpenAI/Claude)
    participant Truto as Truto API

    App->>SDK: initialize(integratedAccountId)
    SDK->>Truto: GET /integrated-account/{id}/tools
    Truto-->>SDK: Return JSON Schema for Lenses tools
    SDK-->>App: Tools ready
    App->>LLM: bindTools(lensesTools)
    App->>LLM: invoke("Restart failed tasks")
    LLM-->>App: tool_call (create_a_lenses_task_restart)
    App->>Truto: POST /proxy/lenses/tasks/restart
    Truto-->>App: 204 No Content
    App->>LLM: Return tool execution result

Step 1: Initialize the Tool Manager

First, you instantiate the Truto SDK and fetch the available tools for your specific Lenses integrated account.

import { ChatOpenAI } from "@langchain/openai";
import { TrutoToolManager } from "truto-langchainjs-toolset";
 
async function runLensesAgent() {
  // 1. Initialize the LLM
  const llm = new ChatOpenAI({
    modelName: "gpt-4o",
    temperature: 0,
  });
 
  // 2. Initialize Truto Tool Manager with your Lenses account ID
  const toolManager = new TrutoToolManager({
    trutoApiKey: process.env.TRUTO_API_KEY!,
    integratedAccountId: "lenses-account-id-123",
  });
 
  // 3. Fetch tools dynamically from Truto
  await toolManager.initialize();
  const tools = toolManager.getTools();
  
  console.log(`Loaded ${tools.length} Lenses tools.`);

Step 2: Bind Tools and Execute

Next, you bind these tools to the LLM and pass the user's prompt.

  // 4. Bind the Lenses tools to the LLM
  const llmWithTools = llm.bindTools(tools);
 
  // 5. Invoke the LLM with a complex DataOps command
  const response = await llmWithTools.invoke([
    {
      role: "system",
      content: "You are a DataOps engineer managing a Lenses Kafka environment. Use the provided tools to query status and resolve issues."
    },
    {
      role: "user",
      content: "Check all Kafka Connect connectors in the production environment. If you find any failed tasks, restart them immediately."
    }
  ]);
 
  // The LLM returns a tool call requesting execution
  console.log("Tool Calls Requested:", response.tool_calls);

Step 3: Handling Execution and API Rate Limits

When your agent executes these tool calls against Lenses via Truto, you must anticipate rate limits.

A factual reality of production integrations: Truto acts as a transparent proxy. It does not automatically retry, throttle, or absorb rate limit errors. If the upstream Lenses API returns an HTTP 429 Too Many Requests, Truto passes that 429 directly back to your agent framework.

However, Truto normalizes the rate limit information into standardized IETF headers across all providers: ratelimit-limit, ratelimit-remaining, and ratelimit-reset. Your agent's execution loop is fully responsible for catching the 429, parsing the ratelimit-reset header, pausing execution, and retrying.

Here is how you handle that execution loop safely:

  // 6. Execute the requested tool calls with Rate Limit awareness
  for (const toolCall of response.tool_calls || []) {
    const tool = tools.find((t) => t.name === toolCall.name);
    if (tool) {
      try {
        console.log(`Executing ${tool.name}...`);
        const result = await tool.invoke(toolCall.args);
        console.log(`Result:`, result);
        
        // You would append this result to the message history 
        // and call the LLM again to continue the chain.
        
      } catch (error: any) {
        if (error.status === 429) {
          // Truto passes the 429 from upstream with standardized headers
          const resetTime = error.headers['ratelimit-reset'];
          console.warn(`Rate limited by Lenses. Wait until ${resetTime} before retrying.`);
          
          // Agent logic to sleep and backoff must go here
          // await sleepUntil(resetTime);
          // retry tool invocation...
        } else {
          console.error(`Tool execution failed: ${error.message}`);
          // Pass error back to the LLM so it can self-correct
        }
      }
    }
  }
}
 
runLensesAgent().catch(console.error);

By feeding the exact failure message back into the model's context, you enable self-correction. If the LLM tried to update an offset while the connector was running, the resulting error will tell the LLM that the state is invalid. The LLM can then reason that it must stop the connector, update the offset, and start it again - entirely autonomously.

The Autonomous DataOps Future

Integrating Lenses with AI agents transforms Kafka from a complex, manually operated infrastructure into an autonomous, self-healing data nervous system. By utilizing Truto's /tools endpoint, you bypass the need to build a custom translation layer for Kafka's nuanced state machine and nested payloads.

Instead of wasting engineering cycles reading Swagger files and maintaining API wrappers, your team can focus on designing the core agent logic and complex workflows. Truto handles the schemas, standardizes the rate limit headers, and ensures that your agent only makes structurally valid requests.