Agent-to-Database: Secure Natural Language Interfaces

Agents that can query databases unlock enormous value—but they also introduce serious security risks. Agent-to-database interfaces translate natural language into SQL, execute queries, and return structured results, all while maintaining strict security boundaries.

This guide covers building secure text-to-SQL interfaces with PostgreSQL, including query validation, SQL injection prevention, and production deployment patterns.

The Text-to-SQL Pipeline

A production text-to-SQL system follows this flow:

User Query → Schema Context → LLM → SQL Generation → Validation → Execution → Results
                                    ↓
                              Query Planning
                              (excludes dangerous operations)

Key Components

1. Schema Context: Database structure available to the LLM
2. Query Generation: LLM produces SQL from natural language
3. Validation Layer: Security checks before execution
4. Safe Execution: Sandboxed query execution
5. Result Formatting: Structured output for the agent

Building the Pipeline

Step 1: Schema Context

Provide the LLM with relevant schema information:

def get_schema_context(tables: list[str]) -> str:
    """Generate schema context for the LLM."""
    context = []
    for table in tables:
        columns = get_columns(table)
        context.append(f"""
Table: {table}
Columns: {', '.join(f'{col.name} ({col.type})' for col in columns)}
Description: {get_table_comment(table)}
""")
    return '\n'.join(context)

Step 2: SQL Generation

Use a structured prompt to generate SQL:

SQL_GENERATION_PROMPT = """You are a SQL expert. Generate a PostgreSQL query for the user's question.

Available schema:
{schema_context}

Rules:
- Only use SELECT queries (no INSERT, UPDATE, DELETE, DROP)
- Use parameterized queries for user inputs
- Include appropriate WHERE clauses for filtering
- Use JOINs to combine related tables
- Return at most 1000 rows unless specified

User question: {question}

Generate a SQL query:"""

Step 3: Query Validation

Validate generated SQL before execution:

import sqlparse
from sqlparse.sql import Statement
from sqlparse.tokens import Keyword, DML

ALLOWED_KEYWORDS = {
    'SELECT', 'FROM', 'WHERE', 'JOIN', 'LEFT', 'RIGHT', 'INNER',
    'ON', 'AND', 'OR', 'NOT', 'IN', 'LIKE', 'BETWEEN', 'IS',
    'NULL', 'ORDER', 'BY', 'GROUP', 'HAVING', 'LIMIT', 'OFFSET',
    'AS', 'DISTINCT', 'COUNT', 'SUM', 'AVG', 'MIN', 'MAX',
    'CASE', 'WHEN', 'THEN', 'ELSE', 'END', 'EXISTS',
}

BLOCKED_KEYWORDS = {
    'INSERT', 'UPDATE', 'DELETE', 'DROP', 'ALTER', 'CREATE',
    'TRUNCATE', 'GRANT', 'REVOKE', 'EXEC', 'EXECUTE',
}

def validate_sql(query: str) -> tuple[bool, str]:
    """Validate a SQL query for safety."""
    parsed = sqlparse.parse(query)[0]

    # Check for blocked operations
    for token in parsed.tokens:
        if token.ttype is Keyword and token.value.upper() in BLOCKED_KEYWORDS:
            return False, f"Blocked keyword: {token.value}"

    # Verify it's a SELECT query
    if not any(t.ttype is DML and t.value.upper() == 'SELECT'
               for t in parsed.tokens):
        return False, "Only SELECT queries are allowed"

    # Check for semicolons (prevent multiple statements)
    if ';' in query:
        return False, "Multiple statements not allowed"

    return True, "Valid"

Step 4: Parameterized Execution

Never concatenate user input into SQL:

import asyncpg

async def execute_query(query: str, params: dict) -> list[dict]:
    """Execute a validated SQL query with parameters."""
    is_valid, error = validate_sql(query)
    if not is_valid:
        raise ValueError(f"Invalid query: {error}")

    # Use parameterized queries
    async with pool.acquire() as conn:
        rows = await conn.fetch(query, *params.values())
        return [dict(row) for row in rows]

Security Patterns

Query Sandboxing

Execute queries in a restricted database role:

-- Create a read-only role for agent queries
CREATE ROLE agent_reader WITH LOGIN;
GRANT CONNECT ON DATABASE mydb TO agent_reader;
GRANT USAGE ON SCHEMA public TO agent_reader;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO agent_reader;

-- Revoke dangerous permissions
REVOKE INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA public FROM agent_reader;

Query Limits

Enforce resource limits:

async def execute_with_limits(query: str, max_rows: int = 1000):
    """Execute query with row limits and timeouts."""
    # Add LIMIT clause if not present
    if 'LIMIT' not in query.upper():
        query = f"{query.rstrip(';')} LIMIT {max_rows}"

    # Set query timeout
    await conn.execute("SET statement_timeout = '5s'")

    return await conn.fetch(query)

Audit Logging

Log all queries for compliance:

import logging
from datetime import datetime

audit_logger = logging.getLogger("agent.database")

async def execute_with_audit(query: str, user_id: str):
    """Execute query with full audit trail."""
    start_time = datetime.now()

    try:
        result = await execute_query(query)
        audit_logger.info(
            f"Query executed | user={user_id} | "
            f"duration={datetime.now() - start_time} | "
            f"rows={len(result)} | query={query}"
        )
        return result
    except Exception as e:
        audit_logger.error(
            f"Query failed | user={user_id} | "
            f"error={e} | query={query}"
        )
        raise

Schema Discovery

Let agents explore database structure safely:

@function_tool
def list_tables() -> list[str]:
    """List all accessible tables in the database."""
    return ["users", "orders", "products"]

@function_tool
def describe_table(table_name: str) -> dict:
    """Get the schema for a specific table."""
    columns = get_columns(table_name)
    return {
        "table": table_name,
        "columns": [
            {"name": col.name, "type": col.type, "nullable": col.nullable}
            for col in columns
        ]
    }

Error Handling

Gracefully handle query failures:

class QueryError(Exception):
    def __init__(self, message: str, query: str = None):
        self.message = message
        self.query = query
        super().__init__(message)

async def safe_query(query: str) -> dict:
    """Execute query with comprehensive error handling."""
    try:
        is_valid, error = validate_sql(query)
        if not is_valid:
            return {"error": f"Query validation failed: {error}"}

        result = await execute_query(query)
        return {"data": result, "row_count": len(result)}

    except asyncpg.exceptions.UndefinedTableError:
        return {"error": "Table not found. Check available tables."}
    except asyncpg.exceptions.DataError as e:
        return {"error": f"Invalid data in query: {e}"}
    except asyncpg.exceptions.PostgresError as e:
        return {"error": f"Database error: {e}"}
    except Exception as e:
        return {"error": f"Unexpected error: {e}"}

Production Deployment

Connection Pooling

import asyncpg

pool = None

async def init_pool():
    global pool
    pool = await asyncpg.create_pool(
        min_size=5,
        max_size=20,
        command_timeout=10,
        # Use read-only connection
        server_settings={"default_transaction_isolation": "read committed"},
    )

Rate Limiting

from collections import defaultdict
import time

class QueryRateLimiter:
    def __init__(self, max_queries_per_minute: int = 30):
        self.max_queries = max_queries_per_minute
        self.queries: dict[str, list[float]] = defaultdict(list)

    def can_query(self, user_id: str) -> bool:
        now = time.time()
        cutoff = now - 60
        self.queries[user_id] = [t for t in self.queries[user_id] if t > cutoff]
        return len(self.queries[user_id]) < self.max_queries

    def record_query(self, user_id: str):
        self.queries[user_id].append(time.time())

Best Practices

1. Never trust LLM output: Always validate before execution
2. Use parameterized queries: Prevent SQL injection
3. Limit database permissions: Read-only for agent access
4. Log everything: Audit trail for compliance
5. Set timeouts: Prevent runaway queries
6. Rate limit: Control query volume
7. Test with production data: Validate against real schemas
8. Monitor query performance: Track slow queries

Next Steps

• Explore Database Indexing for query performance
• Read about Structured Outputs for result validation
• Learn about Agent-to-Database Security in the PostgreSQL docs

Building secure text-to-SQL interfaces requires careful attention to validation, permissions, and monitoring. By implementing these patterns, you can give agents safe access to your data while maintaining security boundaries.