Graph Database Integration — Client Developer Documentation

Introduction

The Graph Database add-on gives agents a unified, pluggable way to work with graphs for knowledge, provenance, and workflow linkage. It supports:

• Remote graph DBs (e.g., JanusGraph via Gremlin, ArangoDB via AQL)
• Local in-memory graph for tests/transient runs
• A common abstract interface for custom backends (e.g., Neo4j)

Agents declare graph capabilities via an Add-On entry in the Subject Specification (similar to how MemoryItem declares memories).

Typical workflow:

1. Declare an AddonItem with addon_type="graph_db" and your chosen backend.
2. Initialize a graph client from agent_sdk.graphs based on that backend.
3. Write vertices/edges, query, and manage graph data.
4. Optionally swap backends (e.g., in-memory for local tests → JanusGraph in prod) without code changes.

---

Registering Graph DB in Subject Specification

Graph DBs are registered using AddonItem.

Data Class Reference

from dataclasses import dataclass, field
from typing import Dict, Any

@dataclass
class AddonItem:
    addon_id: str
    addon_type: str
    addon_backend: str
    addon_config: Dict[str, Any] = field(default_factory=dict)

For graphs: set addon_type="graph_db". addon_backend can be "memory", "janusgraph", "arangodb", "custom" (e.g., your Neo4j client), etc.

---

Minimal Required Configuration

{
  "addon_id": "graphs-primary",
  "addon_type": "graph_db",
  "addon_backend": "janusgraph",
  "addon_config": {
    "host": "janusgraph.default.svc.cluster.local",
    "port": 8182,
    "options": {}
  }
}

---

Field Descriptions

Field	Location	Type	Description
addon_id	root	str	Unique identifier for this add-on.
addon_type	root	str	Must be "graph_db" to enable the graphs module.
addon_backend	root	str	"memory" | "janusgraph" | "arangodb" | "custom" (your own implementation).
addon_config.host / port	config	str/int	Remote endpoint for JanusGraph (Gremlin) or ArangoDB (HTTP).
addon_config.db_name	config	str	(ArangoDB) Database name.
addon_config.username/password	config	str	(ArangoDB) Credentials if required.
addon_config.options	config	dict	Extra backend-specific options (TLS, timeouts, namespaces, etc.).

---

Import and Setup

Choosing a backend at runtime

from agent_sdk.graphs.db import JanusGraphDBClient, ArangoDBClient
from agent_sdk.graphs.local import InMemoryGraphDB

def build_graph_client(addon: dict):
    backend = addon["addon_backend"]
    cfg = addon.get("addon_config", {})

    if backend == "janusgraph":
        return JanusGraphDBClient(host=cfg.get("host", "localhost"),
                                  port=cfg.get("port", 8182))
    elif backend == "arangodb":
        return ArangoDBClient(host=cfg.get("host", "localhost"),
                              port=cfg.get("port", 8529),
                              username=cfg.get("username", "root"),
                              password=cfg.get("password", ""),
                              db_name=cfg.get("db_name", "graph_db"))
    elif backend == "memory":
        return InMemoryGraphDB(directed=cfg.get("directed", True))
    else:
        # e.g., your custom client import/constructor
        raise NotImplementedError(f"Unsupported backend: {backend}")

---

Usage Guide

A. JanusGraph (Gremlin)

Import

from agent_sdk.graphs.db import JanusGraphDBClient

Initialize

client = JanusGraphDBClient(host="localhost", port=8182)

Common operations

v = client.add_vertex("Person", {"name": "Alice", "age": 30})
u = client.add_vertex("Person", {"name": "Bob", "age": 31})
e = client.add_edge(v["id"], u["id"], "KNOWS", {"since": 2024})

read_v = client.get_vertex_by_id(v["id"])
read_e = client.get_edge_by_id(e["id"])

client.delete_edge(e["id"])
client.delete_vertex(v["id"])
client.close()

Custom Gremlin

client.execute_query("g.V().has('Person','name', name)", bindings={"name": "Alice"})

---

B. ArangoDB (AQL)

Import

from agent_sdk.graphs.db import ArangoDBClient

Initialize

client = ArangoDBClient(
    host="localhost", port=8529,
    username="root", password="openSesame",
    db_name="graph_db"
)

Common operations

client.add_document("persons", {"_key": "alice", "name": "Alice"})
doc = client.get_document("persons", "alice")
client.update_document("persons", "alice", {"age": 30})
client.delete_document("persons", "alice")
client.close()

Custom AQL

client.execute_query(
    "FOR p IN persons FILTER p.name == @name RETURN p",
    bind_vars={"name": "Alice"}
)

---

C. Local In-Memory Graph

Import & Init

from agent_sdk.graphs.local import InMemoryGraphDB
graph = InMemoryGraphDB(directed=True)

Common operations

graph.add_vertex("alice")
graph.add_vertex("bob")
graph.add_edge("alice", "bob", relation="KNOWS", since=2023)

graph.find_edges_between("alice", "bob")
graph.find_edges_by_property(relation="KNOWS")

graph.dfs("alice")
graph.find_path("alice", "bob")

graph.remove_edge("alice", "bob")
graph.remove_vertex("alice")

---

Custom Backends

All custom backends must implement the abstract interface:

from abc import ABC, abstractmethod

class GraphDBInterface(ABC):
    @abstractmethod
    def add_vertex(self, *args, **kwargs): ...
    @abstractmethod
    def add_edge(self, *args, **kwargs): ...
    @abstractmethod
    def get_vertex_by_id(self, *args, **kwargs): ...
    @abstractmethod
    def get_edge_by_id(self, *args, **kwargs): ...
    @abstractmethod
    def delete_vertex(self, *args, **kwargs): ...
    @abstractmethod
    def delete_edge(self, *args, **kwargs): ...
    @abstractmethod
    def execute_query(self, *args, **kwargs): ...
    @abstractmethod
    def close(self, *args, **kwargs): ...

Example: Neo4j port

Install

pip install neo4j

Implement

from neo4j import GraphDatabase
from agent_sdk.graphs.abs import GraphDBInterface

class Neo4jDBClient(GraphDBInterface):
    def __init__(self, uri, user, password):
        self.driver = GraphDatabase.driver(uri, auth=(user, password))

    def close(self):
        self.driver.close()

    def _run(self, query, parameters=None):
        with self.driver.session() as session:
            result = session.run(query, parameters or {})
            return [record.data() for record in result]

    def add_vertex(self, label, properties):
        props = ', '.join(f"{k}: ${k}" for k in properties)
        return self._run(f"CREATE (n:{label} {{{props}}}) RETURN n", properties)

    def add_edge(self, out_v, in_v, label, properties=None):
        props = ', '.join(f"{k}: ${k}" for k in (properties or {}))
        q = ("MATCH (a), (b) WHERE id(a)=$out_id AND id(b)=$in_id "
             f"CREATE (a)-[r:{label} {{{props}}}]->(b) RETURN r")
        return self._run(q, {**(properties or {}), "out_id": int(out_v), "in_id": int(in_v)})

    def get_vertex_by_id(self, vertex_id):
        return self._run("MATCH (n) WHERE id(n)=$id RETURN n", {"id": int(vertex_id)})

    def get_edge_by_id(self, edge_id):
        return self._run("MATCH ()-[r]->() WHERE id(r)=$id RETURN r", {"id": int(edge_id)})

    def delete_vertex(self, vertex_id):
        return self._run("MATCH (n) WHERE id(n)=$id DETACH DELETE n", {"id": int(vertex_id)})

    def delete_edge(self, edge_id):
        return self._run("MATCH ()-[r]->() WHERE id(r)=$id DELETE r", {"id": int(edge_id)})

    def execute_query(self, query, bindings=None):
        return self._run(query, bindings)

Use

neo = Neo4jDBClient(uri="bolt://localhost:7687", user="neo4j", password="password")
neo.add_vertex("Person", {"name": "Neo"})
neo.close()

---

Best-Practice Patterns

• Dev/CI: use addon_backend="memory" for fast tests → swap to remote in staging/prod.
• Schema discipline: standardize labels/collections and edge predicates early.
• Observability: wrap execute_query with timing/logging; add retry/backoff for remotes.
• Security: prefer TLS (Gremlin WS / HTTPS) and secrets management for credentials.
• Migration: encapsulate backend choice behind your own factory (see Import and Setup).

---

Quick Subject Spec Example (two add-ons)

{
  "subject_id": "agent-graph-user",
  "subject_type": "agent",
  "addons": [
    {
      "addon_id": "graphs-dev",
      "addon_type": "graph_db",
      "addon_backend": "memory",
      "addon_config": { "directed": true }
    },
    {
      "addon_id": "graphs-prod",
      "addon_type": "graph_db",
      "addon_backend": "janusgraph",
      "addon_config": {
        "host": "janusgraph.default.svc.cluster.local",
        "port": 8182,
        "options": { "tls": false }
      }
    }
  ]
}