SQL (any SQLAlchemy DB)

A single tool that talks to any SQLAlchemy-compatible database — PostgreSQL, MySQL, SQLite, BigQuery, Snowflake, Redshift, MSSQL, Oracle — as long as the driver is installed. Reads are safe by default (the tool refuses anything but SELECT / WITH ... SELECT), mutation is opt-in per tool instance, and a cold start can explore the schema before querying.

TL;DR — SQLTool(url="postgresql://user:pass@host/db").run(ctx, action="query", sql="SELECT 1") returns a markdown table in output.text plus the full rows in metadata.rows. Set allow_writes=True to unlock execute.

When to use

• Finance / revenue reports grounded in live data — action="query" against a warehouse view.
• Schema explorer for an unfamiliar DB — list_tables → describe_table → query.
• Warehouse preflight — schema_summary gives a cheap top-N table / column-count overview before you burn tokens guessing.
• Controlled mutation jobs (backfills, maintenance cleanups) on a dedicated SQLTool(allow_writes=True) instance gated behind human_review.

Setup

SQLTool is not a connector — no CredentialRecord involved. Credentials go into the SQLAlchemy URL (or a pre-built Engine):

pip install 'shipit-agent[sql]'          # pulls sqlalchemy >= 2.0
# and whichever dialect you need, e.g.
pip install 'psycopg[binary]'            # PostgreSQL
pip install 'pymysql'                    # MySQL
pip install 'sqlalchemy-bigquery'        # BigQuery

from shipit_agent.tools.sql import SQLTool

# Option 1 — URL
sql = SQLTool(url="postgresql+psycopg://user:pass@host:5432/db")

# Option 2 — pre-built engine (reuse pooling from the rest of your app)
from sqlalchemy import create_engine
engine = create_engine("bigquery://my-project", echo=False)
sql = SQLTool(engine=engine)

# Option 3 — pick up URL from context state
sql = SQLTool()
# later:
ctx = ToolContext(prompt="demo", state={"sql_url": "sqlite:///:memory:"})

Construction is cheap: SQLAlchemy is lazy-imported on the first run(), not at construction, so building a SQLTool in a factory doesn't force the dependency on unrelated agents.

Safety posture

Exceptions are always caught and surfaced as structured metadata — run() never raises.

Quick example

from shipit_agent.tools.base import ToolContext
from shipit_agent.tools.sql import SQLTool

sql = SQLTool(url="sqlite:///./demo.db", max_rows=200)
ctx = ToolContext(prompt="demo")

out = sql.run(ctx, action="query",
              sql="SELECT region, SUM(arr) AS arr FROM accounts GROUP BY region ORDER BY arr DESC")
print(out.text)
# | region | arr   |
# | ---    | ---   |
# | EU     | 2400  |
# | NA     | 1800  |
# ...

# Full JSON-safe rows are in metadata
for row in out.metadata["rows"][:3]:
    print(row)

As a tool handed to an agent

from shipit_agent import Agent
from shipit_agent.llms import OpenAIChatLLM
from shipit_agent.tools.sql import SQLTool

agent = Agent(
    llm=OpenAIChatLLM(model="gpt-4o-mini"),
    tools=[SQLTool(url="postgresql+psycopg://ro_user:...@analytics-replica/prod")],
    prompt=(
        "You are a revenue analyst. Use the sql tool to answer questions "
        "with live data. Prefer parameterised queries (:name) over string "
        "interpolation."
    ),
)
agent.run("What was Q3 ARR by region? Show the top 5.")

Actions

Five actions live on a single action enum.

query

Read-only SELECT (plus WITH ... SELECT CTEs). Required: sql. Optional: params (dict for :named placeholders).

sql.run(ctx, action="query",
        sql="SELECT id, email FROM users WHERE created_at > :since LIMIT 20",
        params={"since": "2026-01-01"})

Output text renders the first 10 rows as a markdown table for LLM skimming; metadata.rows carries the full (up to max_rows) JSON-safe payload. Decimals become floats, dates become ISO strings, bytes become base64:..., exotic types fall back to str(value). If the result exceeded max_rows, metadata.truncated = True and metadata.total_before_truncation is set.

execute

Mutation (INSERT / UPDATE / DELETE / DDL). Required: sql. Optional: params. Disabled unless the tool was constructed with allow_writes=True; otherwise returns writes_disabled. Runs inside engine.begin() so success commits and SQLAlchemy errors roll back.

write = SQLTool(url=DB_URL, allow_writes=True)
write.run(ctx, action="execute",
          sql="UPDATE users SET plan = :plan WHERE id = :id",
          params={"plan": "pro", "id": 42})
# text: "OK. rowcount=1"

list_tables

No required params. Returns a bullet list of table names plus metadata.tables.

describe_table

Required: table. Optional: schema (qualifier on dialects that have one). Returns per-column name, type, nullable, default, primary_key.

sql.run(ctx, action="describe_table", table="users")

schema_summary

Required: none. Optional: limit (default 50). Returns a top-N table (column_count cols) summary, useful for cheaply introducing a cold warehouse to the LLM.

Error shapes

Related

• Tool catalog — every built-in tool.
• Read file / Grep files — for exploring repos instead of databases.
• Specialists — data / analyst roles that chain schema_summary → describe_table → query.