Databricks SQL

El servidor MCP SQL de Databricks es un servidor MCP gestionado por Azure Databricks que permite a los agentes ejecutar SQL generado por IA sobre tus tablas del Catálogo de Unity para leer y escribir datos, con acceso gobernado por permisos del Catálogo de Unity. Úsalo para construir pipelines de datos a partir de herramientas de codificación con IA. Las consultas se ejecutan de forma asíncrona: el agente llama a la herramienta para iniciar una consulta y luego consulta hasta que la respuesta se completa.

Parámetros _meta

_meta los parámetros son valores de configuración que preestableces en el código de tu agente para establecer el comportamiento del servidor MCP de forma determinista, en lugar de dejar que el LLM los genere dinámicamente en el momento de la llamada a la herramienta. El servidor SQL MCP de Databricks soporta el siguiente _meta parámetro:

Ejemplo: especificar una instancia de SQL Warehouse para consultas SQL de Databricks

En este ejemplo se muestra cómo usar el warehouse_id_meta parámetro para especificar qué almacenamiento SQL ejecuta consultas desde el servidor MCP de SQL de Databricks mediante el SDK de MCP oficial de Python.

En este escenario, deseas:

• Usar un almacenamiento SQL específico para la ejecución de consultas en lugar de permitir que el sistema seleccione uno automáticamente.
• Comprobación del rendimiento coherente mediante el enrutamiento de consultas a un almacenamiento dedicado

Para ejecutar este ejemplo, configure el entorno de Python para el desarrollo de MCP administrado:

Para encontrar su identificador de SQL Warehouse, consulte Conectar a un SQL Warehouse.

# Import required libraries for MCP client and Databricks authentication
import asyncio
from databricks.sdk import WorkspaceClient
from databricks_mcp.oauth_provider import DatabricksOAuthClientProvider
from mcp.client.streamable_http import streamablehttp_client
from mcp.client.session import ClientSession
from mcp.types import CallToolRequest, CallToolResult

async def run_dbsql_tool_call_with_meta():
    # Initialize Databricks workspace client for authentication
    workspace_client = WorkspaceClient()

    # Construct the MCP server URL for DBSQL
    # Replace <workspace-hostname> with your workspace hostname
    mcp_server_url = "https://<workspace-hostname>/api/2.0/mcp/sql"

    # Establish connection to the MCP server with OAuth authentication
    async with streamablehttp_client(
        url=mcp_server_url,
        auth=DatabricksOAuthClientProvider(workspace_client),
    ) as (read_stream, write_stream, _):

        # Create an MCP session for making tool calls
        async with ClientSession(read_stream, write_stream) as session:
            # Initialize the session before making requests
            await session.initialize()

            # Create the tool call request with warehouse_id in _meta
            request = CallToolRequest(
                method="tools/call",
                params={
                    # Tool name for executing SQL queries
                    "name": "execute_sql",

                    # Dynamic arguments - typically provided by your AI agent
                    "arguments": {
                        "query": "SELECT * FROM my_catalog.my_schema.my_table LIMIT 10"
                    },

                    # Meta parameters - specify which warehouse to use
                    "_meta": {
                        "warehouse_id": "a1b2c3d4e5f67890"  # Your SQL warehouse ID
                    }
                }
            )

            # Send the request and get the response
            response = await session.send_request(request, CallToolResult)
            return response

# Execute the async function and get results
response = asyncio.run(run_dbsql_tool_call_with_meta())