execute_sql
Execute SQL queries on Kafka topics across multiple clusters using Lenses.io WebSocket API to manage, explore, transform, and join streaming data.
Instructions
Executes SQL statements/queries using Lenses WebSocket API.
Args: environment: The environment name. sql: The SQL statement/query to execute.
Returns: A list of MessageRecord objects representing the result of the SQL query.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| environment | Yes | ||
| sql | Yes |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- src/lenses_mcp/tools/sql.py:11-24 (handler)The handler function for the 'execute_sql' tool, decorated with @mcp.tool(). It executes the provided SQL query in the specified Lenses environment using the WebSocket API.
@mcp.tool() async def execute_sql(environment: str, sql: str) -> List[Dict[str, Any]]: """ Executes SQL statements/queries using Lenses WebSocket API. Args: environment: The environment name. sql: The SQL statement/query to execute. Returns: A list of MessageRecord objects representing the result of the SQL query. """ endpoint = f"/api/v1/environments/{environment}/proxy/api/ws/v2/sql/execute" return await websocket_client._make_request(endpoint=endpoint, sql=sql) - src/lenses_mcp/server.py:31-31 (registration)Registers the SQL tools (including execute_sql) by calling register_sql(mcp) in the main MCP server setup.
register_sql(mcp) - src/lenses_mcp/server.py:15-15 (registration)Imports the register_sql function used to register the execute_sql tool.
from tools.sql import register_sql