Database Tools
Give your agents a built-in persistent database. Store data in named collections, query with filters, and never worry about database hosting, migrations, or schema setup.
Database tools vs. Knowledge tools
Connic provides two storage systems for agents. They serve different purposes and are often used together.
Stores structured data and finds it by exact field values. Query with operators like $gt, $in, or $andagainst any field in any collection.
Use when:
- Data is structured (orders, users, events, records)
- You need exact lookups, counts, or filters by field
- The agent creates, reads, updates, or deletes records
- Data volume or structure calls for real querying
Stores text and finds it by meaning. Content is turned into vector embeddings, so a query for "cancellation rules" can surface a document titled "return and refund policy".
Use when:
- Searching unstructured text (FAQs, docs, notes)
- Results should be ranked by relevance, not filtered exactly
- The agent needs long-term memory it can recall naturally
- You don't know the exact query in advance
Custom tool wrappers
A common pattern is to wrap the database primitives in domain-specific functions so the agent callssave_order or fetch_orders instead of constructing filters manually. Collection names and field mappings are encoded once in the wrapper, not repeated in the system prompt. For simple agents you can expose the primitives directly in YAML instead. It's up to you.
from connic.tools import db_insert, db_find, db_update, db_delete, db_count
async def save_order(
order_id: str,
customer_email: str,
product: str,
amount: float,
) -> dict:
"""Save a new order to the database."""
return await db_insert("orders", {
"order_id": order_id,
"customer_email": customer_email,
"product": product,
"amount": amount,
"status": "pending",
})
async def fetch_orders(
customer_email: str | None = None,
status: str | None = None,
limit: int = 20,
) -> list:
"""Fetch orders, optionally filtered by customer or status."""
filter_dict = {}
if customer_email:
filter_dict["customer_email"] = customer_email
if status:
filter_dict["status"] = status
result = await db_find(
"orders",
filter=filter_dict,
sort={"amount": -1},
limit=limit,
)
return result["documents"]
async def update_status(order_id: str, new_status: str) -> dict:
"""Update the status of a specific order."""
return await db_update(
"orders",
filter={"order_id": order_id},
update={"status": new_status},
)
async def cancel_order(order_id: str) -> dict:
"""Cancel a pending order."""
return await db_update(
"orders",
filter={"order_id": order_id, "status": "pending"},
update={"status": "cancelled"},
)Use the custom tools in the agent YAML:
name: order-processor
model: gemini/gemini-2.5-flash
description: "Manages orders using the built-in database"
system_prompt: |
You are an order management assistant. Use your tools to
create, look up, update, and delete orders.
tools:
- order_tools.save_order
- order_tools.fetch_orders
- order_tools.update_status
- order_tools.cancel_orderDirect use
For simple agents or prototyping, declare the database tools directly in the YAML and let the agent construct filters itself.
name: db-agent
model: gemini/gemini-2.5-flash
tools:
- db_insert
- db_find
- db_update
- db_delete
- db_count
- db_list_collectionsTool reference
db_find
Query documents from a collection using filters. Supports sorting, pagination, field projection, and distinct value enumeration.
Parameters
| Parameter | Type | Default | Description |
|---|---|---|---|
| collection | str | required | Name of the collection to query |
| filter | dict | {} | Filter dict. See operators below. |
| sort | dict | None | Sort order. 1 = ASC, -1 = DESC. Use underscore prefix for system fields. e.g. {"_created_at": -1} |
| limit | int | 100 | Max documents to return. Hard cap at 1000. |
| skip | int | 0 | Documents to skip. Used for pagination. |
| fields | list | None | Field paths to include. Omit for full documents. e.g. ["id", "status", "amount"] |
| distinct | str | None | If set, return unique values for this field. sort/limit/skip/fields are ignored. |
Returns
Without distinct: {"documents": [...], "count": N}
With distinct: {"values": [...], "count": N}
Each document contains your data fields plus the system fields _id (UUID), _created_at, and _updated_at. System fields always use the underscore prefix, both when reading and when filtering or sorting.
Examples
result = await db_find("orders")
# Returns all orders (up to 100)
result = await db_find("orders", filter={"status": "pending"})
# Returns only pending orders# Sort, paginate, and project
result = await db_find(
"orders",
filter={"amount": {"$gt": 100}},
sort={"amount": -1}, # Highest amount first
limit=20,
skip=40, # Page 3 of 20
fields=["order_id", "customer_email", "amount"],
)
documents = result["documents"]
# Nested field query
result = await db_find("users", filter={"address.city": "Berlin"})
# Multiple conditions
result = await db_find("orders", filter={
"$and": [
{"status": {"$in": ["paid", "shipped"]}},
{"amount": {"$gte": 50}},
]
})
# Distinct values (ignores sort/limit/fields)
result = await db_find("orders", distinct="status")
statuses = result["values"] # ["cancelled", "paid", "pending", "shipped"]Filter operators
System fields use an underscore prefix: _id (UUID), _created_at, _updated_at. Use id (no underscore) for a user-defined field inside your documents. Nested fields use dot notation (e.g. address.city). All values are parameterized server-side and never string-interpolated.
| Operator | Meaning | Example |
|---|---|---|
| {field: value} | Equality (shorthand) | {"status": "active"} |
| $eq | Equal | {"amount": {"$eq": 100}} |
| $ne | Not equal | {"status": {"$ne": "cancelled"}} |
| $gt | Greater than | {"amount": {"$gt": 50}} |
| $gte | Greater than or equal | {"priority": {"$gte": 3}} |
| $lt | Less than | {"score": {"$lt": 0.5}} |
| $lte | Less than or equal | {"age": {"$lte": 30}} |
| $in | Value in list | {"status": {"$in": ["paid", "shipped"]}} |
| $nin | Value not in list | {"status": {"$nin": ["cancelled"]}} |
| $and | All conditions true | {"$and": [{"a": 1}, {"b": 2}]} |
| $or | Any condition true | {"$or": [{"status": "new"}, {"urgent": true}]} |
| $not | Negate condition | {"$not": {"status": "inactive"}} |
| $exists: true | Field exists | {"email": {"$exists": true}} |
| $exists: false | Field does not exist | {"phone": {"$exists": false}} |
| $contains | Array contains value | {"tags": {"$contains": "urgent"}} |
| $regex | Matches regex (case-insensitive) | {"name": {"$regex": "^John"}} |
db_insert
Insert one or more documents into a collection. The collection (and the environment schema) are created automatically on first insert.
Parameters
| Parameter | Type | Default | Description |
|---|---|---|---|
| collection | str | required | Collection name. Must start with a letter; lowercase letters, digits, underscores only. |
| documents | dict | list[dict] | required | A single document or a list of documents. Each document can contain any fields. |
Returns
{"inserted": [...], "inserted_count": N}
Each inserted doc in the list contains your data fields plus system fields _id (auto-generated UUID), _created_at, _updated_at.
Examples
# Insert one document
result = await db_insert("customers", {
"name": "Alice",
"email": "alice@example.com",
"plan": "pro",
})
# result["inserted"][0]["_id"] -> auto-generated UUID
# result["inserted_count"] -> 1# Insert multiple documents at once
result = await db_insert("events", [
{"type": "login", "user": "alice", "ts": "2026-01-01T10:00:00Z"},
{"type": "purchase", "user": "alice", "amount": 49.99},
{"type": "logout", "user": "alice", "ts": "2026-01-01T10:45:00Z"},
])
# result["inserted_count"] -> 3
# Collection is created automatically on first insert
# No setup or schema definition neededdb_update
Update all documents that match a filter. The update dict is merged into each matching document. Fields not mentioned in update are preserved.
Parameters
| Parameter | Type | Default | Description |
|---|---|---|---|
| collection | str | required | Collection name |
| filter | dict | required | Filter dict selecting which documents to update. Empty dict {} updates ALL documents. |
| update | dict | required | Partial document with fields to set. Merged into existing documents. Set a field to null to remove it. |
Returns
{"updated_ids": [...], "updated_count": N}
Examples
# Update a specific document
result = await db_update(
"orders",
filter={"order_id": "ORD-001"},
update={"status": "shipped"},
)
# result["updated_count"] -> 1# Update multiple documents at once
result = await db_update(
"orders",
filter={"status": "pending", "amount": {"$gt": 500}},
update={"status": "priority", "flagged": True},
)
# result["updated_ids"] -> ["uuid1", "uuid2", ...]
# result["updated_count"] -> N
# The update dict is MERGED into existing documents
# Fields not mentioned in update are kept as-isFilter operators
System fields use an underscore prefix: _id (UUID), _created_at, _updated_at. Use id (no underscore) for a user-defined field inside your documents. Nested fields use dot notation (e.g. address.city). All values are parameterized server-side and never string-interpolated.
| Operator | Meaning | Example |
|---|---|---|
| {field: value} | Equality (shorthand) | {"status": "active"} |
| $eq | Equal | {"amount": {"$eq": 100}} |
| $ne | Not equal | {"status": {"$ne": "cancelled"}} |
| $gt | Greater than | {"amount": {"$gt": 50}} |
| $gte | Greater than or equal | {"priority": {"$gte": 3}} |
| $lt | Less than | {"score": {"$lt": 0.5}} |
| $lte | Less than or equal | {"age": {"$lte": 30}} |
| $in | Value in list | {"status": {"$in": ["paid", "shipped"]}} |
| $nin | Value not in list | {"status": {"$nin": ["cancelled"]}} |
| $and | All conditions true | {"$and": [{"a": 1}, {"b": 2}]} |
| $or | Any condition true | {"$or": [{"status": "new"}, {"urgent": true}]} |
| $not | Negate condition | {"$not": {"status": "inactive"}} |
| $exists: true | Field exists | {"email": {"$exists": true}} |
| $exists: false | Field does not exist | {"phone": {"$exists": false}} |
| $contains | Array contains value | {"tags": {"$contains": "urgent"}} |
| $regex | Matches regex (case-insensitive) | {"name": {"$regex": "^John"}} |
db_delete
Delete all documents that match a filter. A non-empty filter is required to prevent accidental full-collection deletion.
Parameters
| Parameter | Type | Default | Description |
|---|---|---|---|
| collection | str | required | Collection name |
| filter | dict | required | Filter dict. Must not be empty. Use _id for system UUID. Use _id with $exists: true to delete all. |
Returns
{"deleted_ids": [...], "deleted_count": N}
Examples
# Delete a document by _id (UUID from db_insert/db_find)
result = await db_delete("orders", {"_id": "550e8400-e29b-41d4-a716-446655440000"})
# result["deleted_count"] -> 1
# Delete all documents in a collection
result = await db_delete("orders", {"_id": {"$exists": True}})
# Delete by a data field
result = await db_delete("orders", {"order_id": "ORD-001"})
# Delete with compound filter
result = await db_delete("logs", {
"$and": [
{"status": "archived"},
{"_created_at": {"$lt": "2025-01-01"}},
]
})
# result["deleted_ids"] -> ["uuid1", ...]Filter operators
System fields use an underscore prefix: _id (UUID), _created_at, _updated_at. Use id (no underscore) for a user-defined field inside your documents. Nested fields use dot notation (e.g. address.city). All values are parameterized server-side and never string-interpolated.
| Operator | Meaning | Example |
|---|---|---|
| {field: value} | Equality (shorthand) | {"status": "active"} |
| $eq | Equal | {"amount": {"$eq": 100}} |
| $ne | Not equal | {"status": {"$ne": "cancelled"}} |
| $gt | Greater than | {"amount": {"$gt": 50}} |
| $gte | Greater than or equal | {"priority": {"$gte": 3}} |
| $lt | Less than | {"score": {"$lt": 0.5}} |
| $lte | Less than or equal | {"age": {"$lte": 30}} |
| $in | Value in list | {"status": {"$in": ["paid", "shipped"]}} |
| $nin | Value not in list | {"status": {"$nin": ["cancelled"]}} |
| $and | All conditions true | {"$and": [{"a": 1}, {"b": 2}]} |
| $or | Any condition true | {"$or": [{"status": "new"}, {"urgent": true}]} |
| $not | Negate condition | {"$not": {"status": "inactive"}} |
| $exists: true | Field exists | {"email": {"$exists": true}} |
| $exists: false | Field does not exist | {"phone": {"$exists": false}} |
| $contains | Array contains value | {"tags": {"$contains": "urgent"}} |
| $regex | Matches regex (case-insensitive) | {"name": {"$regex": "^John"}} |
db_count
Count documents in a collection. Fast - does not load any documents.
Parameters
| Parameter | Type | Default | Description |
|---|---|---|---|
| collection | str | required | Collection name |
| filter | dict | {} | Optional filter. Counts all documents if omitted. |
Returns
{"count": N}
Examples
# Count all documents
result = await db_count("orders")
total = result["count"]
# Count with filter
result = await db_count("orders", {"status": "active"})
active = result["count"]
# Useful for checking before paginating
result = await db_count("events", {"user": "alice"})
pages = (result["count"] + page_size - 1) // page_sizeFilter operators
System fields use an underscore prefix: _id (UUID), _created_at, _updated_at. Use id (no underscore) for a user-defined field inside your documents. Nested fields use dot notation (e.g. address.city). All values are parameterized server-side and never string-interpolated.
| Operator | Meaning | Example |
|---|---|---|
| {field: value} | Equality (shorthand) | {"status": "active"} |
| $eq | Equal | {"amount": {"$eq": 100}} |
| $ne | Not equal | {"status": {"$ne": "cancelled"}} |
| $gt | Greater than | {"amount": {"$gt": 50}} |
| $gte | Greater than or equal | {"priority": {"$gte": 3}} |
| $lt | Less than | {"score": {"$lt": 0.5}} |
| $lte | Less than or equal | {"age": {"$lte": 30}} |
| $in | Value in list | {"status": {"$in": ["paid", "shipped"]}} |
| $nin | Value not in list | {"status": {"$nin": ["cancelled"]}} |
| $and | All conditions true | {"$and": [{"a": 1}, {"b": 2}]} |
| $or | Any condition true | {"$or": [{"status": "new"}, {"urgent": true}]} |
| $not | Negate condition | {"$not": {"status": "inactive"}} |
| $exists: true | Field exists | {"email": {"$exists": true}} |
| $exists: false | Field does not exist | {"phone": {"$exists": false}} |
| $contains | Array contains value | {"tags": {"$contains": "urgent"}} |
| $regex | Matches regex (case-insensitive) | {"name": {"$regex": "^John"}} |
db_list_collections
List all collections in the current environment with document counts and storage sizes. Takes no parameters.
Returns
{"collections": [...], "total": N}
Each collection has name, document_count, size_bytes.
Example
result = await db_list_collections()
for col in result["collections"]:
print(f"{col['name']}: {col['document_count']} docs, {col['size_bytes'] // 1024} KB")
# Result:
# orders: 1543 docs, 200 KB
# customers: 320 docs, 40 KB
# events: 8721 docs, 890 KB