portia.open_source_tools.sql_tool

SQL Tools with pluggable adapter and a default SQLite implementation.

This module provides separate tools for different SQL operations against a database via a pluggable adapter. By default, it ships with a SQLite adapter that can be configured via environment variables or pure JSON config passed at call time.

Available tools:

• RunSQLTool: Execute read-only SQL SELECT queries
• ListTablesTool: List all available tables in the database
• GetTableSchemasTool: Get detailed schema information for specified tables
• CheckSQLTool: Validate SQL queries without executing them

Legacy SQLTool is also available for backward compatibility but is deprecated.

Security note: Only read-only operations are allowed. SQLite authorizer is used to enforce read-only access by denying all write operations (INSERT, UPDATE, DELETE, CREATE, etc.).

SQLAdapter Objects

class SQLAdapter()

Abstract adapter interface for SQL databases (read-only).

run_sql

def run_sql(query: str) -> list[dict[str, Any]]

Execute a read-only query and return rows as list of dicts.

list_tables

def list_tables() -> list[str]

List available table names.

get_table_schemas

def get_table_schemas(tables: list[str]) -> dict[str, list[dict[str, Any]]]

Return column schemas for the given tables.

check_sql

def check_sql(query: str) -> dict[str, Any]

Check if a query would run successfully (read-only).

SQLiteConfig Objects

@dataclass
class SQLiteConfig()

Configuration for SQLite connections.

Attributes:

• db_path - Path to the SQLite database file, or ":memory:" for in-memory.

SQLiteAdapter Objects

class SQLiteAdapter(SQLAdapter)

SQLite adapter using read-only URI mode where possible.

__init__

def __init__(config: SQLiteConfig) -> None

Initialize the adapter with the given configuration.

run_sql

def run_sql(query: str) -> list[dict[str, Any]]

Execute the read-only SQL query and return rows as dicts.

list_tables

def list_tables() -> list[str]

Return a list of user tables in the database.

get_table_schemas

def get_table_schemas(tables: list[str]) -> dict[str, list[dict[str, Any]]]

Return PRAGMA table_info for each table in tables.

check_sql

def check_sql(query: str) -> dict[str, Any]

Check the query by executing an EXPLAIN; return ok True/False with error.

BaseSQLToolArgs Objects

class BaseSQLToolArgs(BaseModel)

Base arguments for SQL tools.

Either provide config via:

• environment variables, or
• the optional config_json string with adapter-specific config (pure JSON)

BaseSQLTool Objects

class BaseSQLTool(Tool[Any])

Base SQL tool with shared adapter functionality.

Use SQLiteAdapter by default. Configure via env or config_json:

• SQLITE_DB_PATH: path to sqlite database (e.g., /tmp/db.sqlite). If not set, defaults to :memory:

__init__

def __init__(adapter: SQLAdapter | None = None, **kwargs: Any) -> None

Initialize the tool with an optional adapter (defaults to SQLite).

RunSQLArgs Objects

class RunSQLArgs(BaseSQLToolArgs)

Arguments for running SQL queries.

RunSQLTool Objects

class RunSQLTool(BaseSQLTool)

Execute read-only SQL SELECT queries against a database.

run

def run(_: ToolRunContext, **kwargs: Any) -> list[dict[str, Any]]

Execute the SQL query and return results.

ListTablesArgs Objects

class ListTablesArgs(BaseSQLToolArgs)

Arguments for listing database tables.

ListTablesTool Objects

class ListTablesTool(BaseSQLTool)

List all available tables in the database.

run

def run(_: ToolRunContext, **kwargs: Any) -> list[str]

List all tables in the database.

GetTableSchemasArgs Objects

class GetTableSchemasArgs(BaseSQLToolArgs)

Arguments for getting table schemas.

GetTableSchemasTool Objects

class GetTableSchemasTool(BaseSQLTool)

Get detailed schema information for specified tables.

run

def run(_: ToolRunContext, **kwargs: Any) -> dict[str, list[dict[str, Any]]]

Get schema information for the specified tables.

CheckSQLArgs Objects

class CheckSQLArgs(BaseSQLToolArgs)

Arguments for checking SQL query validity.

CheckSQLTool Objects

class CheckSQLTool(BaseSQLTool)

Check if a SQL query is valid without executing it.

run

def run(_: ToolRunContext, **kwargs: Any) -> dict[str, Any]

Check the validity of the SQL query.