Ak Dev New Knowledgebase Integration

Adding a New Knowledge Base Integration

Use this skill to add a new knowledge base backend under ak-py/src/agentkernel/knowledgebase/.

This applies when you are integrating any durable source not already covered by:

• ChromaManager (semantic vector search)
• Neo4jManager (graph relationships)
• StarburstManager (read-only SQL via Trino)
• OKFManager (Open Knowledge Format markdown bundles, over a local directory or S3)

If the new source is documents addressed by path, do not start from scratch: compose a DocumentStore instead (step 4a).

Prerequisites

• Understand architecture and contribution patterns from .agents/skills/ak-dev-architecture/SKILL.md
• Understand existing knowledge base APIs:
  • ak-py/src/agentkernel/knowledgebase/base.py (the ABC, the capability gates, schema())
  • ak-py/src/agentkernel/knowledgebase/model.py (KnowledgeCapabilities, record typing)
  • ak-py/src/agentkernel/knowledgebase/errors.py (the error hierarchy)
  • ak-py/src/agentkernel/knowledgebase/knowledgebuilder.py (tool gating)
  • ak-py/src/agentkernel/knowledgebase/document.py and store/base.py (document-shaped backends)
  • ak-py/src/agentkernel/knowledgebase/okf/manager.py (the reference document backend)
• Have provider credentials and a local/dev test instance for the target backend

Step-by-Step

1. Create Backend Module

Create a new file:

ak-py/src/agentkernel/knowledgebase/<backend>.py

Use lowercase file names (for example qdrant.py, milvus.py, elasticsearch.py).

2. Subclass KnowledgeBase

Implement a concrete class that extends KnowledgeBase:

from typing import Any, Iterable, List, Mapping

from .base import KnowledgeBase, Record
from .model import KnowledgeCapabilities


class MyBackendManager(KnowledgeBase):
    def __init__(self, name: str = "", description: str | None = None, **kwargs):
        # Declare only what this backend actually supports: an undeclared operation
        # raises KnowledgeCapabilityError rather than returning an empty result.
        super().__init__(
            capabilities=KnowledgeCapabilities(
                kinds=["vector"],
                search=True,
                search_mode="semantic",
                writable=True,
            ),
            name=name,
        )
        self.name = name
        self.description = description or "my backend"
        self._client = None
        self.connect(**kwargs)

    @property
    def backend_name(self) -> str:
        return self.name if self.name else "mybackend"

    def connect(self, **kwargs) -> None:
        # Initialize backend client and verify connectivity.
        self._client = ...

    def write(self, records: Iterable[Record], **kwargs) -> None:
        for record in records:
            text = str(record.get("text", "")).strip()
            metadata = dict(record.get("metadata", {}))
            if not text:
                continue
            # Persist text + metadata using backend-native API.
            self._client.store(text=text, metadata=metadata)

    def search(self, query: str, limit: int = 3, **kwargs) -> List[Mapping[str, Any]]:
        rows = self._client.search(query=query, limit=limit)
        return [{"text": row["text"], "metadata": row.get("metadata", {})} for row in rows]

    def get_description(self) -> str:
        return f"{self.backend_name}: {self.description}"

super().__init__(capabilities=...) is required — the base takes no zero-argument form. Pass name too so a rejected declaration names your backend rather than the class.

2a. Declare the Right Capability Set

backend_name, connect and get_description are the only abstract members. Each retrieval and write operation is optional, and the declaration decides which ones you must implement:

Two fields are descriptive rather than gating: kinds is an open taxonomy (vector, structured, graph, document, …) and search_mode ("semantic" / "lexical") is advisory — it does not imply search, and declaring search does not require it.

Rules KnowledgeBase.__init__ enforces:

• At least one of search / query / fetch / browse / writable must be True.
• query=True requires a query_language ("cypher", "sql", …), and a query_language requires query=True.

KnowledgeBase.validate_capabilities(capabilities, subject) is a static method, so a declaration can be checked without constructing a backend — which is how the contract suite exercises it.

Never implement read(). There is no such method on the ABC, and one defined on your subclass is never called. The routing it used to do lives in the read_kb tool, which reaches query() when you declared query and search() otherwise — which is what lets one tool serve every backend.

search_kb's gate is the same shape as the other two: any backend declaring search. For a search-only backend it therefore reaches the same search() that read_kb does. That redundancy is deliberate — the two tools differ in what they promise, not in what they reach — and it is what lets a backend with no query language still advertise relevance retrieval by name.

If the backend derives its own schema, declare derives_schema=True and return a non-empty mapping from _derived_schema(); the contract suite fails a backend that declares one without the other. schema() writes capabilities last and unoverridably, so a deployment cannot contradict the declaration through add_schema().

3. Respect the Record Contract

All reads must return normalized rows compatible with KnowledgeBuilder:

• {"text": str, "metadata": dict}

All writes accept the same shape through write(records=[...]).

If the provider result is not in this shape, normalize inside search() / query() / fetch() / browse().

model.py names the shape in two TypedDicts — KnowledgeRecord (text, metadata) and KnowledgeMetadata (id, source, title, kind, trust, stale, links). They are documentation-only. Record = Mapping[str, Any] stays the annotation on every signature, nothing validates them at runtime, and a backend is free to carry keys beyond the conventional set. Use them to decide what to name your metadata, not as a schema to enforce.

One key is load-bearing: a backend declaring fetch must put a comma-free id in every record's metadata. format_results() prefixes each line with it so the agent can feed a result straight back into fetch_kb, and fetch_kb splits its ids argument on ,. KnowledgeBaseContract asserts this; nothing checks it at runtime.

4. Define Backend Constraints Explicitly

If the backend is read-only, say so in the declaration and implement nothing:

super().__init__(
    capabilities=KnowledgeCapabilities(kinds=["structured"], query=True, query_language="sql", writable=False),
    name=name,
)

KnowledgeBase.write() then raises KnowledgeCapabilityError(backend_name, "write") on its own, and KnowledgeBuilder gates write_kb before it ever reaches you. Do not override write() to raise — that only duplicates the base — and never silently ignore writes. The same holds for every other operation: leave undeclared ones unimplemented rather than raising by hand.

Raise into the hierarchy in errors.py, not out of it:

• KnowledgeError is the base every knowledge-base failure subclasses.
• KnowledgeCapabilityError is an undeclared operation. It is deliberately not a NotImplementedError: that would be indistinguishable from an unimplemented abstract method, whereas this is a declaration mismatch the tool boundary is expected to catch and explain.
• KnowledgePathError is a path that escapes a store namespace or is unusable as an identity.

Finding nothing is not a failure. A read matching no records returns []; a document that cannot be parsed is skipped with a diagnostic. Reserve exceptions for failures of the machinery.

4a. Document-Shaped Backends: Compose a DocumentStore

If your records are documents at paths, subclass DocumentKnowledgeBase rather than KnowledgeBase and let a DocumentStore supply the bytes. That splits the two axes — storage and representation — so one backend serves the same collection from a local directory in development and an object store in production, with no code change. OKFManager (okf/manager.py) is the reference implementation.

class MyDocBackend(DocumentKnowledgeBase):
    def __init__(self, store: DocumentStore, name: str = "") -> None:
        super().__init__(
            store=store,
            capabilities=KnowledgeCapabilities(kinds=["document"], search=True, fetch=True, browse=True, writable=True),
            name=name,
        )

What the base gives you, and what it does not:

• Writability folds with and. The store's writable is intersected with your declaration, so the more restrictive side wins: a read-only store beats a backend willing to write. This is why capabilities are per instance, not per class.
• _read_document(path) maps absence to None, not an exception — FileNotFoundError and other OSErrors (a browse record fed back into fetch reaches open() on a directory) become a logged warning and an empty answer. KnowledgePathError deliberately propagates.
• Containment is the store's obligation, not yours. .. segments, absolute paths, and symlinks escaping a local root are refused on access and skipped during traversal. Decide per operation what to do with the refusal — dropping one path out of a fetch list and refusing a whole write are different answers — but never re-implement the check.
• Configuration is a single string. DocumentStore.from_uri() accepts a bare path, file://, s3://bucket/prefix, and python:pkg.mod.ClassName for a bring-your-own store. Take a DocumentStore in your constructor; do not take a path and build the store yourself.

5. Add Robust Connection Handling

• Validate required configuration fields in connect()
• Fail fast with clear ValueError messages for missing settings
• Add reconnection logic when the provider client commonly drops stale sessions
• Implement close() if the backend has sockets, cursors, or open sessions

6. Add Optional Dependencies

Update ak-py/pyproject.toml with a new optional dependency group:

[project.optional-dependencies]
mybackend = [
    "provider-sdk>=x.y.z",
]

Keep dependency groups narrow and provider-specific.

Two cases the template does not cover:

• A pure-Python backend may need no extra at all. The OKF backend adds none — pyyaml is a core dependency. Do not invent an extra for a group that would be empty.
• The extra belongs where the import is. boto3 arrives through S3DocumentStore and the existing aws extra, not through the backend that composes it.

Then decide whether the backend belongs in knowledgebase/__init__.py's _LAZY_EXPORTS. The rule is the reason ChromaManager / Neo4jManager / StarburstManager are deliberately not exported: a module that imports its SDK at module import would make that SDK a hard requirement the moment an agent touched the name, even lazily. Export the name only if importing your module pulls no optional dependency; otherwise leave callers to import it from its own module, and say so in its docstring. If you do export it, add the TYPE_CHECKING mirror entry too.