delightful-carpenter-30895
09/13/2025, 3:10 AMtall-airline-16756
09/13/2025, 3:10 AMtall-airline-16756
09/13/2025, 3:10 AMpg
, mysql2
, or sqlite3
. You can securely pass credentials through job metadata or environment variables to keep everything safe.
If you'd like, I’d be more than happy to guide you through implementing a specific database—say MongoDB—with either Python or Node.js in a LiveKit agent. Just let me know how I can support you!
Sources: Tool definition and use | LiveKit Docs | complex-agents/personal_shopper/database.py | Tool definition and use | LiveKit Docs | Function calling with Voice Agents | LiveKit Docs | External data and RAG | LiveKit Docsdelightful-carpenter-30895
09/13/2025, 3:13 AMtall-airline-16756
09/13/2025, 3:14 AMpymongo
for MongoDB or `psycopg2`/`asyncpg` for PostgreSQL.
Here’s a simple example for MongoDB to get you started:
from livekit.agents import function_tool
import pymongo
@function_tool
async def query_user_data(user_id: str):
client = pymongo.MongoClient("<mongodb://your-host:27017/>")
db = client["your_db"]
user = db.users.find_one({"id": user_id})
return str(user) or "User not found"
And here’s how you might do it for PostgreSQL using `psycopg2`:
from livekit.agents import function_tool
import psycopg2
@function_tool
async def get_user_orders(user_id: str):
conn = psycopg2.connect(
host="your-postgres-host",
database="your_db",
user="your_user",
password="your_password"
)
cursor = conn.cursor()
cursor.execute("SELECT * FROM orders WHERE user_id = %s", (user_id,))
results = cursor.fetchall()
cursor.close()
conn.close()
return str(results)
If you're looking to handle asynchronous operations, I recommend considering asyncpg
for PostgreSQL or wrapping blocking calls in a thread pool. Please feel free to reach out if you'd like any further guidance or help with the implementation—I’m here to support you every step of the way!
Sources: External data and RAG | LiveKit Docs | External data and RAG | LiveKit Docs | complex-agents/nutrition-assistant/agent.py | examples/drive-thru/drivethru_agent.py