Query a SQLite Database

#Goal

Use a configured Daita agent to inspect and query an existing SQLite database without giving the model a general SQL connection.

#Prepare the Agent

Run the interactive terminal once to create the agent and configure its model:

bash

daita

Then attach an existing database:

bash

daita attach sales sqlite /absolute/path/sales.db --source-name Sales
daita sources sales

Daita opens the file read-only and commits its discovered tables, columns, indexes, and relationships to the agent's catalog.

#Ask a Question

python

import asyncio
from daita import Agent
 
 
async def main() -> None:
    agent = await Agent.open("sales")
    try:
        source = await agent.resolve_source("Sales")
        result = await agent.run(
            "Show completed order count and revenue by region for the last 90 days.",
            source_id=source.id,
        )
 
        print(result.final_text)
        print("run:", result.run_id)
        print("conversation:", result.conversation_id)
 
        follow_up = await agent.run(
            "Which region changed the most compared with the prior 90 days?",
            conversation_id=result.conversation_id,
        )
        print(follow_up.final_text)
 
        transcript = await agent.transcript(result.run_id)
        for message in transcript.messages:
            print(message)
    finally:
        await agent.close()
 
 
asyncio.run(main())

The model first works from cataloged structure, then can call data_query_sqlite with one validated read-only statement. The SQL connector enforces catalog scope and result bounds before returning rows to the model.

#Continue the Conversation

The second run() in the example reuses the returned conversation ID. The source selected for the first turn remains pinned to that conversation, and the follow-up receives a bounded projection of prior completed runs.

#Inspect What Happened

agent.transcript() returns the exact record of the selected run. If the SQLite schema changes later, reopen the agent and refresh before querying again:

python

agent = await Agent.open("sales")
try:
    source = await agent.resolve_source("Sales")
    await agent.refresh_source(source.id)
finally:
    await agent.close()

Continue with SQLite, Catalog, and Conversations.