agent-kernel: Skill for Claude Code

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 collections.abc import Iterable, Mapping
from typing import Any

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}"