Custom Database Connectors

📋 Planned

Documentation Under Construction

This guide will show you how to implement a custom database connector by implementing the SqlRunner interface, allowing Vanna to connect to any SQL database.

Planned Content

✓ The SqlRunner interface explained
✓ Complete example implementation (e.g., for DuckDB or ClickHouse)
✓ Handling connection management and pooling
✓ Error handling and retries
✓ Testing your custom runner
✓ Row-level security considerations
✓ Registering your custom runner with the agent

Want to contribute or suggest improvements? Open an issue on GitHub

The SqlRunner Interface

When complete, this will show:

from vanna.core.tool import SqlRunner, RunSqlToolArgs, ToolContext
import pandas as pd

class MyCustomDatabaseRunner(SqlRunner):
    """Connect to YourDatabase"""

    def __init__(self, connection_string: str):
        self.connection_string = connection_string
        # Initialize your database client

    async def run_sql(
        self,
        args: RunSqlToolArgs,
        context: ToolContext
    ) -> pd.DataFrame:
        """Execute SQL query and return results as DataFrame"""
        # 1. Get user context for row-level security
        user = context.user

        # 2. Apply any filters based on user permissions
        filtered_query = self.apply_user_filters(args.query, user)

        # 3. Execute query with your database client
        results = await self.execute_query(filtered_query)

        # 4. Return as pandas DataFrame
        return pd.DataFrame(results)

Use Cases

This guide will be useful for:

Uncommon databases: DuckDB, ClickHouse, TimescaleDB, etc.
Custom data sources: APIs, data warehouses with special auth
Enhanced security: Custom row-level filtering logic
Special requirements: Connection pooling, query rewriting, logging