Internal IT Support Agent

Notion’s built-in AI keyword-searches. It returns the document you asked about and stops. It can’t tell you why a decision was made, who influenced it, or whether it’s been superseded by something newer. HydraDB is different. It doesn’t just store vectors - it builds a living context graph. Every memory is parsed, enriched, and connected to other memories. When your agent calls POST /query, it doesn’t just get semantically similar chunks. It gets the most useful context for that exact query - weighted by recency, relevance, relationships, and historical usage patterns.

All code in this cookbook is real. Base URL is https://api.hydradb.com. Get your API key at app.hydradb.com.

Prerequisites

Required knowledge: Python basics, REST APIs, environment variables
Required tools:

HydraDB API key
Python 3.11 or 3.12 (python --version)
pip install hydradb-sdk

What You’ll Build

By the end of this cookbook, you’ll be able to:

Ingest Notion pages, Confluence docs, and Slack threads into a unified HydraDB workspace
Answer “why did we decide X?” by retrieving decision context across linked documents
Batch-upload documents with verified indexing before any search query
Build a conversational interface grounded in your team’s actual knowledge

How HydraDB Works

Before writing code, understand the three primitives you’ll use throughout this cookbook:

Database - your workspace. All data is isolated per database. Think of it as your “company” in HydraDB. Create one per application.
Memory - any unit of context: a Notion page, a Confluence doc, a Slack thread, a user preference. HydraDB automatically chunks, embeds, and connects memories into a context graph.
Search - the retrieval call your agent makes before acting. HydraDB’s search runs a multi-stage pipeline: metadata filtering → graph traversal → semantic retrieval → personalized ranking.

LongMemEvals search accuracy: 90%

Comparison: Traditional RAG vs HydraDB Search

Feature	Traditional RAG	HydraDB search
Search method	Vector search - nearest neighbors only	Multi-stage: intent → graph → semantic → rank
Scale	Context collapses at 10M+ documents	Petabyte scale, sub-second latency
Personalization	No personalization across users	Personalized per user via `user_name`
Context awareness	No relationship or decision awareness	Context graph links docs, people, decisions
Accuracy	Constant tuning of ranking heuristics	90% accuracy on LongMemEvals

Step 01 - Create a Database

Every HydraDB workspace starts with a database. Create one for your knowledge base - it provides complete data isolation and multi-tenant support out of the box. Endpoint: POST /databases - Create your workspace

Bash

curl -X POST 'https://api.hydradb.com/databases' \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"database": "notion-ai-workspace"}'

Python

import os, time, uuid
from hydra_db import HydraDB

API_KEY   = os.environ["HYDRA_DB_API_KEY"]
DATABASE_ID = "notion-ai-workspace"

client = HydraDB(token=API_KEY)

# Create database - idempotent, safe to re-run
resp = client.tenants.create(database=DATABASE_ID)
print("Database:", resp)

Collections for teams: Use collection to isolate data by department. Engineering, Sales, HR each get their own namespace within your database - no configuration needed, just pass the ID on upload.

Step 02 - Upload Knowledge Memories

HydraDB automatically parses, chunks, embeds, and connects your content into a context graph. You don’t manage embeddings or vector indexes. You just upload.

Notion Connector

Fetch pages from Notion, format them into HydraDB’s app source structure, and batch upload. HydraDB builds the context graph automatically - no edge creation needed.

Batch limit: Max 20 sources per request. Wait 1 second between batches to respect rate limits.

import json, time
from notion_client import Client

notion = Client(auth=os.environ["NOTION_TOKEN"])

def extract_text(page_id: str) -> str:
    """Extract plain text from all rich_text blocks on a page."""
    blocks = notion.blocks.children.list(block_id=page_id)["results"]
    lines  = []
    for b in blocks:
        rt   = b.get(b["type"], {}).get("rich_text", [])
        text = "".join(r["plain_text"] for r in rt)
        if text:
            lines.append(text)
    return "\n\n".join(lines)


def upload_batch(sources: list, collection: str = None) -> list:
    """Upload up to 20 sources. Returns list of IDs."""
    if collection:
        for s in sources:
            s["collection"] = collection
    result = client.context.ingest(
        database=DATABASE_ID,
        app_knowledge=json.dumps(sources),
    )
    return [r.id for r in (result.data.results or []) if r.id]


def ingest_notion_database(database_id: str, collection: str = None) -> list:
    pages   = notion.databases.query(database_id=database_id)["results"]
    batch   = []
    all_ids = []

    for page in pages:
        props      = page["properties"]
        title_prop = props.get("Name", props.get("Title", {}))
        title_arr  = title_prop.get("title", [])
        title      = title_arr[0]["plain_text"] if title_arr else "Untitled"
        text       = extract_text(page["id"])
        author     = page["created_by"]["id"]

        batch.append({
            "id":        page["id"],
            "title":     title,
            "type":      "notion_page",                    # required
            "timestamp": page["last_edited_time"],         # required ISO
            "content":   {"text": text},
            "url":       f"https://notion.so/{page['id'].replace('-','')}",
            "metadata": {
                "author": author,
                "tags":   ["notion", "knowledge"],
            },
        })

        if len(batch) == 20:
            all_ids += upload_batch(batch, collection)
            batch = []
            time.sleep(1)  # required 1-second interval between batches

    if batch:
        all_ids += upload_batch(batch, collection)
    return all_ids

Confluence Connector

Confluence pages follow the same upload format. Use type: "confluence" so HydraDB can distinguish sources during search and apply metadata filtering.

from atlassian import Confluence
from bs4 import BeautifulSoup

conf = Confluence(
    url=os.environ["CONFLUENCE_URL"],
    username=os.environ["CONFLUENCE_USER"],
    password=os.environ["CONFLUENCE_TOKEN"],
    cloud=True,
)

def ingest_space(space_key: str, collection: str = None) -> list:
    pages   = conf.get_all_pages_from_space(
        space_key, expand="body.storage,history,version"
    )
    batch   = []
    all_ids = []

    for page in pages:
        html = page["body"]["storage"]["value"]
        text = BeautifulSoup(html, "html.parser").get_text("\n\n")

        batch.append({
            "id":        page["id"],
            "title":     page["title"],
            "type":      "confluence",                    # required
            "timestamp": page["version"]["when"],         # required ISO
            "content":   {"text": text},
            "metadata": {
                "author": page["history"]["createdBy"]["accountId"],
                "tags":   ["confluence", space_key.lower()],
            },
        })

        if len(batch) == 20:
            all_ids += upload_batch(batch, collection)
            batch = []
            time.sleep(1)

    if batch:
        all_ids += upload_batch(batch, collection)
    return all_ids

Verify Indexing

After uploading, always verify that HydraDB has fully processed and indexed your content before running search queries. Endpoint: GET /context/status?ids=ID&database=DATABASE - Check indexing status

def poll_until_indexed(id: str, timeout: int = 120, interval: int = 3):
    """Poll until indexed, errored, or timeout."""
    deadline = time.time() + timeout
    while time.time() < deadline:
        result = client.context.status(
            database=DATABASE_ID,
            ids=id,
        )
        items  = result.data.statuses or []
        status = items[0].indexing_status if items else None
        if status == "completed":
            return True
        if status == "errored":
            raise RuntimeError(f"Indexing failed for {id}")
        time.sleep(interval)
    raise TimeoutError(f"Indexing timed out for {id}")

def verify_all(ids: list):
    for fid in ids:
        poll_until_indexed(fid)

Step 03 - Add User Memories

Beyond documents, HydraDB stores user memories - preferences, habits, and patterns that personalize search per user. Set infer: true to let HydraDB extract implicit signals from text. Set infer: false to store facts verbatim. Endpoint: POST /context/ingest - Add a user memory

Bash

curl -X POST 'https://api.hydradb.com/context/ingest' \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "memories": [{
      "text": "Alice prefers concise bullet-point answers and always wants source links",
      "user_name": "alice",
      "infer": true
    }],
    "database": "notion-ai-workspace",
    "collection": "user-alice",
    "upsert": true
  }'

Python

def add_user_memory(
    user_name: str,
    preference: str,
    collection: str = None,
    infer: bool = True,
) -> dict:
    """
    Store a user memory/preference in HydraDB.
    Uses the SDK-spec body format with the memories[] array wrapper.
    """
    payload = {
        "memories": [{
            "text":      preference,
            "user_name": user_name,
            "infer":     infer,
        }],
        "database": DATABASE_ID,
        "upsert":    True,
    }
    if collection:
        payload["collection"] = collection

    return client.context.ingest(
    type='memory',**payload)


# Example usage
add_user_memory(
    user_name="alice",
    preference="Alice prefers concise bullet-point answers and always wants source links",
    collection="user-alice",
    infer=True,
)

After a few interactions, HydraDB builds a behavioral model per user. Alice’s search results will be ranked and formatted differently from Bob’s - automatically, without any extra code.

Step 04 - Search Context

This is the call your agent makes before answering any question. POST /query runs HydraDB’s full multi-stage pipeline and returns ranked, contextually relevant chunks - including graph context showing relationships between entities. Endpoint: POST /query - Retrieve agent context

def recall_context(
    query: str,
    collection: str     = None,
    max_results: int    = 10,
    alpha: float        = 0.8,    # FIX: single definition, no duplicate
    recency_bias: float = 0.3,
    graph_context: bool = True,
) -> dict:
    """
    Full search - searches knowledge base (documents).
    To also retrieve user memories, call /query separately.
    collection: scope to a specific workspace or user namespace.
    mode: "thinking" enables personalised ranking.
    graph_context: true enables cross-document entity linking.
    """
    payload = {
        "database":     DATABASE_ID,
        "query":         query,
        "max_results":   max_results,
        "mode":          "thinking",
        "graph_context": graph_context,
        "alpha":         alpha,       # FIX: single key only
        "recency_bias":  recency_bias,
    }
    if collection:
        payload["collection"] = collection

    return client.query(**payload)
    # Response shape:
    # data["chunks"]        - ranked context chunks with relevancy_score
    # data["graph_context"] - entity paths and chunk_relations


# Example
context = recall_context(
    "Why did we migrate from MySQL to Postgres?",
    collection="workspace",
)
for chunk in (context.data.chunks or []):
    print(f"[{chunk.relevancy_score or 0:.2f}] {chunk.source_title or ''}")
    print((chunk.chunk_content or "")[:200])

Step 05 - Search and Answer Generation

For conversational, AI-generated answers, first retrieve context with POST /query. Then pass the returned chunks and sources into your application-layer LLM prompt. Key parameters: alpha (0-1, balance semantic vs keyword bm25), recency_bias (0-1, prefer newer content), and graph_context. Endpoint: POST /query - retrieved context for app-layer answer generation

def ask_workspace(
    question: str,
    collection: str     = None,
    source_filter: str  = None,
    alpha: float        = 0.5,
    recency_bias: float = 0.3,
    max_results: int    = 10,
) -> dict:
    """
    Retrieve workspace context for a question.
    Returns chunks, sources, and graph_context. Generate the final answer in your app layer.
    """
    payload = {
        "database":     DATABASE_ID,
        "query":         question,
        "max_results":   max_results,
        "graph_context": True,
        "mode":          "thinking",
        "alpha":         alpha,
        "recency_bias":  recency_bias,
    }
    if collection:
        payload["collection"] = collection
    if source_filter:
        payload["metadata_filters"] = {"source_type": source_filter}

    return client.query(**payload)


def build_context(result) -> str:
    return "\n\n".join(
        chunk.chunk_content or ""
        for chunk in (result.data.chunks or [])
    )


# Usage examples
result = ask_workspace("Why did we choose Postgres over MySQL?")
print(build_context(result)[:1000])

# Follow-up questions call search again and your app includes any prior chat turns in the LLM prompt.
result2 = ask_workspace("What were the tradeoffs they considered?")
print(build_context(result2)[:1000])

# Filter to only Notion pages
notion_result = ask_workspace(
    "What's in our engineering RFC library?",
    source_filter="notion_page",
)

Maintaining conversation context: Store chat history in your application and include relevant prior turns in your LLM prompt. HydraDB returns retrieval context; your app owns the final answer and conversation state.

Step 06 - Slack Interface

Expose your knowledge base as a Slack bot. When a user mentions @wiki, the bot calls ask_workspace() to retrieve context, then your application can pass that context to an LLM for the final Slack response.

from slack_bolt import App

app = App(token=os.environ["SLACK_BOT_TOKEN"])

# Persist session IDs so conversation memory works across turns
user_sessions: dict = {}

def get_session(user_id: str) -> str:
    """Return or create a persistent session ID for this Slack user."""
    return user_sessions.setdefault(user_id, str(uuid.uuid4()))

@app.event("app_mention")
def handle_mention(event, client):
    slack_user = event["user"]
    question   = event["text"].split(">", 1)[-1].strip()

    # Acknowledge immediately so Slack doesn't time out
    msg = client.chat_postMessage(
        channel=event["channel"],
        thread_ts=event["ts"],
        text="_Searching your workspace..._",
    )

    session_id = get_session(slack_user)
    result     = ask_workspace(question)

    chunks = result.data.chunks or []
    answer = "\n\n".join(c.chunk_content or "" for c in chunks[:3]) or "No results found."
    sources = result.data.sources or []

    if sources:
        links  = "\n".join(f"• {s.title or ''}" for s in sources[:3])
        answer += f"\n\n*Sources:*\n{links}"

    client.chat_update(
        channel=event["channel"],
        ts=msg["ts"],
        text=answer,
    )

if __name__ == "__main__":
    from slack_bolt.adapter.socket_mode import SocketModeHandler
    SocketModeHandler(app, os.environ["SLACK_APP_TOKEN"]).start()

API Reference

All endpoints used in this cookbook. Base URL: https://api.hydradb.com. Header: Authorization: Bearer YOUR_API_KEY

Database management

POST /databases - Create workspace - idempotent

{ "database": "notion-ai-workspace" }

Upload app sources (Notion, Slack, Confluence…)

POST /context/ingest - Max 20/call, 1s between batches

[{
  "id":        "page-uuid",
  "title":     "RFC-041 Database Migration",
  "type":      "notion_page",                    // required
  "timestamp": "2024-09-01T08:00:00Z",          // required ISO
  "content":   { "text": "We chose Postgres because..." },
  "url":       "https://notion.so/...",
  "metadata": {
    "author": "[email protected]",
    "tags":   ["rfc", "database"]
  }
}]

Upload a single file (PDF / DOCX)

POST /context/ingest - Single file with database as form field

curl -X POST 'https://api.hydradb.com/context/ingest' \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -F "documents=@/path/to/document.pdf" \
  -F "database=notion-ai-workspace"

Verify processing

GET /context/status?ids=ID&database=YOUR_DATABASE - Poll until status = “completed”

Full search

POST /query - Searches knowledge base - returns chunks + graph_context

{
  "database":     "notion-ai-workspace",
  "collection": "workspace",
  "query":         "Why did we choose Postgres?",
  "max_results":   10,
  "mode":          "thinking",
  "graph_context": true,
  "alpha":         0.8,
  "recency_bias":  0.3
}

Search for answer generation

POST /query - retrieved context for app-layer answer generation

{
  "database":        "notion-ai-workspace",
  "query":            "Why did we choose Postgres?",
  "max_results":      10,
  "mode":             "thinking",
  "graph_context":    true,
  "alpha":            0.5,
  "recency_bias":     0.3,
  "metadata_filters": { "source_type": "notion_page" }
}

Add user memory

POST /context/ingest - memories[] array wrapper - consistent SDK format

{
  "memories": [{
    "text":      "Alice prefers bullet-point responses",
    "user_name": "alice",
    "infer":     true
  }],
  "database":     "notion-ai-workspace",
  "collection": "user-alice",
  "upsert":        true
}

Search user memories

POST /query - user_name key - consistent across all calls

{
  "database":     "notion-ai-workspace",
  "collection": "user-alice",
  "user_name":     "alice",
  "query":         "How should I format answers for this user?"
}

Delete memory

DELETE /context - Remove stale or incorrect memory data with type: "memory" and request.ids

Benchmarks

HydraDB leads LongMemEvals with 90% search accuracy. Compared to a naive RAG pipeline over the same 12,400-document corpus:

Query type	Naive RAG	HydraDB	Delta
Factual lookup queries	81% search	90% search	+11%
“Why did we…” decision queries	34% search	79% search	+132%
Stale doc surface rate	41% of results	7% of results	−83%
P95 query latency	220ms	under 200 ms	Sub-second

Benchmark methodology. Figures are based on internal HydraDB testing. For the formal benchmark paper see research.hydradb.com/hydradb.pdf. Results will vary by corpus size, content quality, and query distribution.

​Prerequisites

​What You’ll Build

​How HydraDB Works

​Comparison: Traditional RAG vs HydraDB Search

​Step 01 - Create a Database

​Bash

​Python

​Step 02 - Upload Knowledge Memories

​Notion Connector

​Confluence Connector

​Verify Indexing

​Step 03 - Add User Memories

​Bash

​Python

​Step 04 - Search Context

​Step 05 - Search and Answer Generation

​Step 06 - Slack Interface

​API Reference

​Database management

​Upload app sources (Notion, Slack, Confluence…)

​Upload a single file (PDF / DOCX)

​Verify processing

​Full search

​Search for answer generation

​Add user memory

​Search user memories

​Delete memory

​Benchmarks

Prerequisites

What You’ll Build

How HydraDB Works

Comparison: Traditional RAG vs HydraDB Search

Step 01 - Create a Database

Bash

Python

Step 02 - Upload Knowledge Memories

Notion Connector

Confluence Connector

Verify Indexing

Step 03 - Add User Memories

Bash

Python

Step 04 - Search Context

Step 05 - Search and Answer Generation

Step 06 - Slack Interface

API Reference

Database management

Upload app sources (Notion, Slack, Confluence…)

Upload a single file (PDF / DOCX)

Verify processing

Full search

Search for answer generation

Add user memory

Search user memories

Delete memory

Benchmarks