Turso Edge Database

SQLite at the edge with embedded replicas for microsecond-latency AI agents

Turso provides SQLite-compatible databases at the edge with embedded replicas, delivering microsecond reads and global distribution for your AI agents.

import os
from praisonaiagents import Agent, MemoryConfig

agent = Agent(
    name="Edge Agent",
    instructions="You are a fast assistant with edge persistence.",
    memory=MemoryConfig(
        db={
            "database_url": os.environ["TURSO_DATABASE_URL"],
            "auth_token": os.environ["TURSO_AUTH_TOKEN"],
        },
    ))
agent.start("Remember that I prefer concise answers")

The user shares a preference; Turso edge storage persists it for fast reads on the next turn.

How It Works

Quick Start

1. Get Turso Credentials

   1. Install Turso CLI: curl -sSfL https://get.tur.so/install.sh | bash
   2. Create database: turso db create mydb
   3. Get credentials:

   turso db show mydb --url
   turso db tokens create mydb
   

   Set environment variables:

   export TURSO_DATABASE_URL="libsql://mydb-user.turso.io"  
   export TURSO_AUTH_TOKEN="eyJhbGci..."
   
2. Create Edge Agent

   Deprecated import. from praisonai.db.adapter import TursoDB and the top-level from praisonai import DB / PraisonAIDB / PraisonDB now emit DeprecationWarning — as of PR #4595 the top-level aliases route through praisonai.db so the warning fires consistently. Use the db(...) factory instead:

   from praisonaiagents import db
   
   store = db(database_url="libsql://mydb-user.turso.io", auth_token="eyJ...")
   

   from praisonaiagents import Agent, MemoryConfig, db
   
   store = db(
       database_url="libsql://mydb-user.turso.io",
       auth_token="eyJhbGci...",
   )
   
   agent = Agent(
       name="Edge Agent",
       instructions="You are a fast AI assistant with edge persistence.",
       memory=MemoryConfig(db=store),
   )
   
   # Microsecond reads from local replica
   result = agent.start("I need responses with minimal latency")
   print(result)
   
3. Test Edge Performance

   import time
   
   # Time a conversation round-trip
   start = time.perf_counter()
   result = agent.start("What's the quickest database option?")
   end = time.perf_counter()
   
   print(f"Response time: {(end - start) * 1000:.1f}ms")
   print(f"Agent: {result}")
   # Typically 1-5ms for local reads
   

---

Installation

pip

# Install with Turso/libSQL support
pip install "praisonai[turso]"

Environment Variables

# Required for remote sync
export TURSO_DATABASE_URL="libsql://mydb-user.turso.io"
export TURSO_AUTH_TOKEN="eyJhbGci..."

# Optional
export OPENAI_API_KEY="sk-..."

---

Connection Modes

praisonai persistence doctor auto-detects libsql://… URLs as the turso backend — no --*-backend flag needed. See Backend Auto-Detection.

Embedded Replica (Recommended)

Local SQLite file syncs with remote Turso server:

from praisonai.persistence.conversation.turso import TursoConversationStore

store = TursoConversationStore(
    url="libsql://mydb-user.turso.io",
    auth_token="eyJhbGci...",
    local_path="agent_replica.db",  # Local SQLite file
    sync_on_write=True  # Sync to remote after writes
)

# Reads are microsecond-fast from local file
# Writes sync to global edge network

Remote Only

Direct connection to Turso edge:

store = TursoConversationStore(
    url="libsql://mydb-user.turso.io", 
    auth_token="eyJhbGci...",
    local_path=None  # No local replica
)

# All operations go to remote - higher latency but always consistent

Local Only

Pure local SQLite for development:

store = TursoConversationStore(
    url=None,  # No remote sync
    local_path="dev_agent.db"
)

# Fastest possible - no network calls
# Perfect for development and testing

---

Configuration Options

Option	Type	Default	Description
url	str	None	Turso database URL (libsql://...)
auth_token	str	None	Turso authentication token
local_path	str	"praisonai_turso.db"	Local SQLite replica file
sync_on_write	bool	True	Sync to remote after writes
auto_create_tables	bool	True	Create conversation tables

---

Usage Patterns

Using the db Factory

from praisonaiagents import Agent, MemoryConfig, db

# Reads TURSO_DATABASE_URL and TURSO_AUTH_TOKEN from the environment
store = db(database_url="libsql://mydb-user.turso.io", auth_token="eyJ...")
agent = Agent(name="Edge Agent", memory=MemoryConfig(db=store))

Legacy import. from praisonai.db.adapter import TursoDB still works but now emits DeprecationWarning (PR #4595). Prefer the db(...) factory above.

Multi-Region Setup

# Primary region
primary_db = TursoDB(
    database_url="libsql://primary-us-east.turso.io",
    auth_token="eyJ...",
    local_path="primary_replica.db"
)

# Secondary region  
secondary_db = TursoDB(
    database_url="libsql://secondary-eu-west.turso.io", 
    auth_token="eyJ...",
    local_path="secondary_replica.db"
)

# Use closest database for lowest latency
agent = Agent(name="Global Agent", memory={"db": primary_db})

Full Lifecycle with Edge Performance

import os
import time
from praisonai import ManagedAgent, LocalManagedConfig
from praisonai.db.adapter import TursoDB
from praisonaiagents import Agent, MemoryConfig

# Phase 1: Create edge agent
db = TursoDB(
    database_url=os.environ["TURSO_DATABASE_URL"],
    auth_token=os.environ["TURSO_AUTH_TOKEN"],
    local_path="edge_agent_replica.db"
)

managed = ManagedAgent(
    provider="local",
    db=db, 
    config=LocalManagedConfig(
        model="gpt-4o-mini",
        name="Turso Edge Agent",
        system="You are a high-performance AI assistant with edge database."
    )
)

agent = Agent(name="User", backend=managed)

# Store facts with performance timing
start = time.perf_counter()
result1 = agent.run("Remember: I'm building a real-time multiplayer game")
sync_time = time.perf_counter() - start
print(f"Write + sync time: {sync_time * 1000:.1f}ms")
print(f"Agent: {result1}")

# Fast read from local replica
start = time.perf_counter() 
result2 = agent.run("What am I building?")
read_time = time.perf_counter() - start
print(f"Read time: {read_time * 1000:.1f}ms")  # Typically < 5ms
print(f"Agent: {result2}")

# Phase 2: Save session
session_data = managed.save_ids()
del agent, managed, db

# Phase 3: Resume from replica (instant)
db2 = TursoDB()  # Uses environment variables
managed2 = ManagedAgent(provider="local", db=db2)
managed2.resume_session(session_data["session_id"]) 

agent2 = Agent(name="User", backend=managed2)
start = time.perf_counter()
result3 = agent2.run("What type of game requires real-time performance?")
resume_time = time.perf_counter() - start
print(f"Resume read time: {resume_time * 1000:.1f}ms")
print(f"Resumed Agent: {result3}")

---

Turso-Specific Features

Manual Sync Control

Control when local replica syncs to remote:

from praisonai.persistence.conversation.turso import TursoConversationStore

store = TursoConversationStore(
    url="libsql://mydb.turso.io",
    auth_token="eyJ...",
    sync_on_write=False  # Manual sync control
)

# Batch multiple operations
store.create_session(session1)
store.add_message(session_id, message1)
store.add_message(session_id, message2)

# Sync all changes at once
store._sync()

Database Scaling

Turso automatically scales to handle traffic:

# Monitor database metrics
turso db show mydb --metrics

# View recent usage
turso db show mydb --usage

Global Replication

Create replicas in multiple regions:

# Add replica in Europe
turso db replicate mydb --region fra

# Add replica in Asia  
turso db replicate mydb --region nrt

# List all replicas
turso db show mydb --replicas

# Connect to closest replica automatically
agent = Agent(
    name="Global Agent",
    memory=MemoryConfig(
        db={
            "database_url": "libsql://mydb-user.turso.io",  # Closest replica chosen automatically
            "auth_token": "eyJ..."
        },
    ))

---

Best Practices

Optimize for Edge Performance

Use embedded replicas for fastest reads:

from praisonai.db.adapter import TursoDB

# Embedded replica: microsecond reads, durable writes
db = TursoDB(
    local_path="fast_agent_replica.db",  # Local SSD storage
    sync_on_write=True  # Immediate durability
)

# For highest performance, batch writes
store = TursoConversationStore(sync_on_write=False)
# ... multiple operations ...
store._sync()  # Single sync operation

Handle Network Partitions

Embedded replicas continue working during network outages:

# Agent keeps working even if network fails
agent = Agent(
    name="Resilient Agent",
    instructions="You work offline with local replica.",
    memory=MemoryConfig(
        db=TursoDB(local_path="offline_replica.db"),
    ))

# Reads always work from local file
# Writes queue for next sync when network returns

Monitor Replica Size

Local replicas grow with conversation history:

import os

replica_path = "agent_replica.db"
if os.path.exists(replica_path):
    size_mb = os.path.getsize(replica_path) / 1024 / 1024
    print(f"Replica size: {size_mb:.1f}MB")
    
    # Archive old conversations if replica gets large
    if size_mb > 100:  # 100MB threshold
        # Implement archiving logic
        pass

Use Multiple Databases

Separate different data types for optimal performance:

# Separate DBs for different purposes
conversation_db = TursoDB(
    database_url="libsql://conversations.turso.io",
    local_path="conversations.db"
)

analytics_db = TursoDB(
    database_url="libsql://analytics.turso.io", 
    local_path="analytics.db"
)

# Agent uses conversation DB, analytics service uses analytics DB
agent = Agent(name="Chat Agent", memory={"db": conversation_db})

---

Environment Variables

Variable	Required	Format	Example
TURSO_DATABASE_URL	✅	libsql://db-user.turso.io	libsql://mydb-user.turso.io
TURSO_AUTH_TOKEN	✅	JWT token	eyJhbGciOiJFZERTQS...
OPENAI_API_KEY	✅	Set in environment	Your provider API key

---

Performance Comparison

Operation	Turso Replica	Remote Only	Local Only
Read	< 5ms	20-100ms	< 1ms
Write	5-50ms	20-150ms	< 1ms
Consistency	Eventually	Strong	None
Offline	✅	❌	✅

---

Troubleshooting

Sync Failures

If replica sync fails:

# Check connection and retry sync manually
try:
    store._sync()
    print("Sync successful")
except Exception as e:
    print(f"Sync failed: {e}")
    # Continue with local operations

Token Expiration

Turso tokens can expire. Regenerate them:

# Create new token
turso db tokens create mydb

# Update environment variable
export TURSO_AUTH_TOKEN="new-token-here"

Replica File Permissions

Ensure local replica file is writable:

import os
replica_path = "agent_replica.db"

# Check if directory is writable
if not os.access(os.path.dirname(replica_path) or ".", os.W_OK):
    print("Directory not writable!")

---

Related

Cloud Databases Overview

Compare all cloud database providers

Local SQLite

Pure local SQLite for development