talk2sql 0.1.0
A lean, extensible MCP server framework for natural language to SQL interactions
License: MIT
Requires Python: >=3.10
Keywords: mcp, sql, llm, natural-language, database, model-context-protocol

# Talk2SQL

> A lean, extensible MCP server framework for natural language to SQL interactions.

[![PyPI version](https://badge.fury.io/py/talk2sql.svg)](https://pypi.org/project/talk2sql/)
[![Tests](https://img.shields.io/badge/tests-169%20passing-brightgreen)](https://github.com/talk2sql/talk2sql)
[![Python 3.10+](https://img.shields.io/badge/python-3.10+-blue.svg)](https://www.python.org/downloads/)

**Talk2SQL** provides an abstract base class for building [Model Context Protocol (MCP)](https://modelcontextprotocol.io) servers that enable LLMs to interact with relational databases through natural language.

## Features

✨ **Zero Abstract Methods** - All methods have sensible defaults
🔒 **Security First** - SQL injection detection, query type validation, table ACLs
🚀 **Database Agnostic** - Works with any DB-API 2.0 compliant connection
📦 **Lean Dependencies** - Only `fastmcp` and `sqlparse`
⚡ **Smart Caching** - LRU cache with TTL for metadata
🎯 **Few-Shot Learning** - JSON-based example configuration
🛠️ **Extensible** - Configuration injection over inheritance

## Quick Start

### Installation

```bash
pip install talk2sql
```

### Basic Usage

```python
from talk2sql import Talk2SQLBase, GuardrailConfig
from talk2sql.dialects import POSTGRES
import psycopg2

# Create database connection
conn = psycopg2.connect(
    host="localhost",
    database="mydb",
    user="user",
    password="pass"
)

# Create Talk2SQL server
server = Talk2SQLBase(
    connection=conn,
    dialect=POSTGRES,
    config_path="./db_config.json",  # Optional: few-shot examples
    guardrails=GuardrailConfig(
        allowed_query_types=["SELECT"],  # Read-only by default
        blocked_tables=["sensitive_data", "audit_logs"],
        max_rows=500,
        query_timeout_seconds=30
    ),
    cache_ttl=600  # Cache metadata for 10 minutes
)

# Create and run MCP server
mcp = server.create_mcp_server()
mcp.run()
```

### Configuration File (Optional)

Create a JSON file to provide few-shot examples and business context:

```json
{
  "version": "1.0",
  "examples": [
    {
      "question": "How many users are active?",
      "sql": "SELECT COUNT(*) FROM users WHERE status = 'active'",
      "description": "Count active users"
    }
  ],
  "global_context": "E-commerce database. Monetary amounts in cents. Timestamps in UTC.",
  "table_context": {
    "users": "Core accounts table. status: 'A' = active, 'I' = churned.",
    "orders": "One row per order. total_amount in cents."
  }
}
```

## Supported Databases

Talk2SQL includes built-in dialect configurations for:

- **PostgreSQL** (`talk2sql.dialects.POSTGRES`)
- **MySQL** (`talk2sql.dialects.MYSQL`)
- **Oracle** (`talk2sql.dialects.ORACLE`)

### Custom Dialects

```python
from talk2sql import Talk2SQLBase, DialectConfig

MY_CUSTOM_DB = DialectConfig(
    name="custom_db",
    tables_query="SELECT name, schema, description FROM sys.tables",
    columns_query="SELECT col, type, nullable FROM sys.columns WHERE tbl = '{table}'",
    timeout_command="SET TIMEOUT {timeout}"
)

server = Talk2SQLBase(connection=my_connection, dialect=MY_CUSTOM_DB)
```

## Architecture

Talk2SQL follows the **Open/Closed Principle**:

```
┌─────────────────────────────────────────────────────────────┐
│                      Talk2SQLBase                           │
├─────────────────────────────────────────────────────────────┤
│ Constructor Injection:                                      │
│   ├── connection: Any         (Duck-typed cursor protocol)  │
│   ├── dialect: DialectConfig  (SQL templates for metadata)  │
│   ├── guardrails: GuardrailConfig                           │
│   ├── config_path: Path | None (JSON for examples/context)  │
│   └── cache_ttl: int = 300    (Metadata cache seconds)      │
├─────────────────────────────────────────────────────────────┤
│ MCP Tools:                                                  │
│   ├── search_tables   - Search table names/descriptions     │
│   ├── get_table_schema - Get columns and types for table    │
│   └── run_query       - Execute validated SQL query         │
├─────────────────────────────────────────────────────────────┤
│ MCP Resources:                                              │
│   ├── examples://      - All few-shot examples              │
│   ├── context://global - Global business context            │
│   └── context://tables - All table-specific contexts        │
└─────────────────────────────────────────────────────────────┘
```

### Five Pillars

| Pillar | Implementation |
|--------|----------------|
| **Metadata Discovery** | Query system catalogs with LRU caching |
| **Security/Guardrails** | Query type validation, table ACL, SQL injection detection |
| **Query Execution** | Duck-typed connections, parameterized queries, row limits |
| **Few-Shot Examples** | JSON configuration exposed as MCP resource |
| **Custom Prompts** | Table/global context in JSON, exposed as MCP resources |

## Security

Talk2SQL includes comprehensive security features:

### SQL Injection Detection

- Multiple statement detection (`;` separator)
- Comment injection (`--`, `/* */`)
- UNION-based attacks
- Tautology patterns (`OR 1=1`)
- Time-based injection (`SLEEP`, `WAITFOR`)
- Hex encoding detection

### Query Validation

```python
guardrails = GuardrailConfig(
    allowed_query_types=["SELECT"],  # Block writes by default
    blocked_tables=["secrets", "audit_logs"],  # Table blacklist
    max_rows=1000,  # Prevent large result sets
    query_timeout_seconds=30  # Database-level timeout
)
```

## Examples

### MySQL Example

```python
from talk2sql import Talk2SQLBase
from talk2sql.dialects import MYSQL
import mysql.connector

conn = mysql.connector.connect(
    host="localhost",
    database="mydb",
    user="user",
    password="pass"
)

server = Talk2SQLBase(connection=conn, dialect=MYSQL)
mcp = server.create_mcp_server()
mcp.run()
```

### Oracle Example

```python
from talk2sql import Talk2SQLBase
from talk2sql.dialects import ORACLE
import oracledb

conn = oracledb.connect(
    user="user",
    password="pass",
    dsn="localhost:1521/ORCLPDB"
)

server = Talk2SQLBase(connection=conn, dialect=ORACLE)
mcp = server.create_mcp_server()
mcp.run()
```

### Custom Guardrails

```python
from talk2sql import Talk2SQLBase, GuardrailConfig

class MyServer(Talk2SQLBase):
    def apply_guardrails(self, sql: str) -> GuardrailResult:
        # Add custom validation logic
        result = super().apply_guardrails(sql)

        if "SENSITIVE_COLUMN" in sql.upper():
            return GuardrailResult(
                allowed=False,
                reason="Access to SENSITIVE_COLUMN is prohibited",
                query_type=result.query_type,
                referenced_tables=result.referenced_tables
            )

        return result
```

## API Reference

### Talk2SQLBase

**Methods:**

- `search_tables(query: str) -> List[TableMeta]` - Search for tables by name/description
- `get_table_schema(table_name: str) -> TableSchema` - Get detailed schema for a table
- `apply_guardrails(sql: str) -> GuardrailResult` - Validate SQL query
- `execute(sql: str) -> QueryResult` - Execute SQL with validation
- `create_mcp_server(name: str = "talk2sql") -> FastMCP` - Create MCP server instance
- `invalidate_cache() -> int` - Clear metadata cache

**Properties:**

- `dialect: DialectConfig` - Current dialect configuration
- `guardrails: GuardrailConfig` - Current guardrail configuration
- `cache_stats: Dict[str, Any]` - Cache statistics
- `examples: List[Example]` - Loaded few-shot examples
- `global_context: str` - Global business context
- `table_contexts: Dict[str, str]` - Table-specific contexts

## Development

### Running Tests

```bash
# Install with dev dependencies
pip install -e ".[dev]"

# Run tests
pytest

# Run with coverage
pytest --cov=talk2sql --cov-report=html
```

### Project Structure

```
talk2sql/
├── __init__.py              # Public exports
├── base.py                  # Talk2SQLBase class
├── config.py                # DialectConfig, GuardrailConfig
├── models.py                # Data models
├── guardrails.py            # SQL validation
├── cache.py                 # LRU cache with TTL
├── dialects/
│   ├── __init__.py
│   ├── postgres.py
│   ├── mysql.py
│   └── oracle.py
└── exceptions.py            # Custom exceptions
```

## Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

## License

MIT License - see LICENSE file for details

## Links

- [PyPI Package](https://pypi.org/project/talk2sql/)
- [Model Context Protocol](https://modelcontextprotocol.io)
- [FastMCP Documentation](https://github.com/jlowin/fastmcp)

---

Built with ❤️ using the Model Context Protocol