Develop: Cultivators

How to author, test, and ship a Cultivator Kind. For the conceptual model (Kinds vs bindings, the Scope Validator, Connections) see Concepts: Cultivators; this page is the engineering guide. The in-repo counterpart is docs/cultivators/authoring.md; the worked example to copy is packages/datahub/src/datahub/templates/examples/hr_skills/cultivators/skill_extractor.py (an example Kind — its HR domain is incidental).

A Kind is a BaseCultivator subclass that consumes ingest events and returns Statement Proposals. The behavior lives in PR-reviewed code; per-tenant activation and narrowing live in tenant_cultivators rows managed over the API — never hardcode tenant specifics into a catalog Kind.

1. Pick the directory

The framework derives the expected namespace from the module path at registration and fails the import on a mismatch — a bespoke Kind physically cannot be bound by any other tenant (404 kind_not_found). This is hard isolation, unlike agents which are scope-gated.

2. Subclass BaseCultivator

from datahub.cultivators import (
    BaseCultivator,
    CultivatorContext,
    CultivatorOutput,
    StatementProposal,
)


class PerfSignals(BaseCultivator):
    name = "perf_signals"
    version = "0.1"
    tenant_namespace = "acme_corp"          # None for catalog Kinds
    accepts = (("transcript", "meeting"),)  # (source_type, data_type) pairs
    allowed_scope_patterns = ("managers_of:*:read",)   # write ceiling
    runtime_scopes = ("users:read", "context:performance:read")  # read ceiling
    default_model = "anthropic/claude-sonnet-4-6"
    description = "Extracts performance signals from meeting transcripts."
    run_timeout_seconds = 120.0             # optional; default 120s

    async def should_process(self, event) -> bool:
        # Cheap, deterministic, NO LLM. False ⇒ run recorded as 'skipped'.
        return len(event.payload.get("participants", [])) >= 2

    async def reflect(self, event, ctx: CultivatorContext) -> CultivatorOutput:
        ...


PerfSignals.register()

Attribute reference (packages/datahub/src/datahub/cultivators/base.py):

• namesnake_case, unique within its namespace slot.
• versionbump on any behavior change (prompt, nodes, default model). cultivator_runs.kind_version attributes every statement to the version that wrote it.
• acceptsthe (source_type, data_type) pairs from /v1/ingest events this Kind dispatches on.
• allowed_scope_patternsthe Scope Grant Catalog: the only scope patterns proposals may carry. Tenants can narrow it (scope_grant_overrides), never broaden.
• runtime_scopeswhat the Kind can read from DataHub during reflection, bound through RLS before reflect runs. Least privilege per vertical.
• requires_connectionsoptional tuple of ConnectionUse(name, tools, on_missing) for investigator Kinds that call external MCP servers. on_missing="skip" (default) skips the run when the tenant hasn't configured the connection; "degrade" proceeds without it.

3. Implement reflect

Convention: a small LangGraph with three nodes — fetch_context (cheap I/O, no LLM), reflect (LLM calls via litellm.acompletion(model=ctx.model, ...), each wrapped in ctx.tracer.log_generation(...)), propose (assemble proposals). Copy the shape from skill_extractor.py.

Everything the Kind touches comes through ctx:

• ctx.tools.subjects.resolve(identifier)resolve a person reference to a verified dh_users UUID.
• ctx.tools.graphium.search_context / recall_context — read prior memory under the Kind's read ceiling.
• ctx.tools.mcp.call(server, tool, args)external calls for investigator Kinds, authorized against both the tenant Connection's allowed_tool_patterns and the Kind's declared tools.
• ctx.model, ctx.prompt_append, ctx.tracer, ctx.run_id, ctx.tenant_id.

Return value:

return CultivatorOutput(
    proposals=[
        StatementProposal(
            text="Maria led the incident review for the Q2 outage.",
            family_slug="performance",
            genus_slug="evidence",
            required_scopes=[f"managers_of:{subject.uid}:read"],
            about_person_ids=[subject.uid],
            memory_type="event",
            source_ref="transcript://meet/2026-06-02",
        ),
    ],
    reasoning_trace="...",   # ≤16KB, lands in the run audit
)

Or skip explicitly: CultivatorOutput(proposals=[], skipped=True, skip_reason="no participants resolved"). One proposal per atomic fact; the framework applies the same 0.999-cosine dedup as manual ingest.

4. Respect the scope discipline

Every proposal's required_scopes is validated against the Kind's catalog, then the tenant's narrowing, then subject existence. The four rejection codes (malformed_scope, scope_outside_catalog, tenant_narrow_violation, unknown_subject) are tallied on the run with only a SHA-256 of the text — rejected content is never stored.

Hard rule: the LLM never constructs scope strings. Python code builds them from trusted tool output — ctx.tools.subjects.resolve(...) UUIDs — never from raw event payload fields. The validator catches forged UUIDs; building scopes only from resolver output catches honest mistakes earlier.

5. Wire registration

Call YourKind.register() at module bottom, then side-effect import the module from the template's cultivators/__init__.py:

# templates/custom/acme_corp/cultivators/__init__.py
from datahub.templates.custom.acme_corp.cultivators import perf_signals  # noqa: F401

Duplicate registration with a different class raises at import — deployments fail fast rather than shadow silently.

6. Test

Mandatory per the platform constitution; AAA pattern. Copy the skill_extractor test shape.

• Unit (packages/datahub/tests/unit/templates/.../cultivators/test_<kind>.py) — should_process branches; reflect happy path with mocked tools + mocked litellm asserting proposal count and required_scopes shape; skip branches; defense-in-depth check that scopes derive from resolver UUIDs only.
• Integration (packages/datahub/tests/integration/templates/.../cultivators/test_<kind>_end_to_end.py) — against real Postgres with stubbed LLM: assert the cultivator_runs row, the landed statements carry acaso:provenanceRun, the filter skip writes nothing, and a hallucinated subject finalizes as rejected_validation.

uv run pytest packages/datahub/tests/unit/templates/ -k perf_signals

7. Bind it for the tenant

After deploy, the Kind appears in GET /v1/cultivators (for bespoke Kinds: only for the matching tenant). New bindings start disabled — enabling is a deliberate flip.

curl -sS -X POST https://<host>/v1/cultivators \
  -H "Authorization: Bearer $ADMIN_KEY" -H "Content-Type: application/json" \
  -d '{"kind": "perf_signals", "enabled": true}'

Investigator Kinds additionally need the tenant's Connections registered once via POST /v1/connections (connections:write), including the secret_ref that the worker resolves to an env var at call time. Operate and audit runs via GET /v1/cultivators/runs and the per-run Langfuse deep link — see Workflows: Manage cultivators.

Pitfalls

• Don't call ingest_context from a Kind — return proposals; the toolset excludes it on purpose.
• Don't read current_user_id() — cultivators have no impersonated user.
• Don't bypass ctx.tools.mcp with direct HTTP/SDK calls — that skips allowlist enforcement and integrations_used accounting.
• Don't build subject scopes from payload fields — resolver UUIDs only.
• Don't swallow exceptions in reflect — let the runner record the failure with its cause in cultivator_runs.error.

Ship checklist

• Directory matches the namespace; register() at import; template __init__ imports the module.
• allowed_scope_patterns and runtime_scopes reviewed as ceilings, least privilege both sides.
• Unit + integration tests in AAA shape; all make gates green.
• Tenant binding created (and Connections, for investigators); first runs inspected in the runs audit.
• version set; bumped on every behavior change thereafter.