Extending Hive

Every extension point with copy-paste code examples.

Unified memory (daemon + tools)

When memory.unified: true (default), agents have one semantic store at .hive/memory/<agent_id>/ shared by the daemon pursuit loop, goal generation, MemoryToolkit, and KnowledgeToolkit. Legacy JSON KV files under .hive/agent_memory/ migrate once on first access.

flowchart LR
    MT["MemoryToolkit<br/>memory_set / memory_get"]
    KT["KnowledgeToolkit<br/>knowledge_store"]
    SM["SemanticMemory<br/>.hive/memory/agent_id/"]
    DA["Daemon pursuit<br/>PersistentMemory recall"]
    GG["Goal generation<br/>memory_snippets"]

    MT --> SM
    KT --> SM
    DA --> SM
    GG --> SM

Standalone SDK tasks can still attach PersistentMemory(hive_dir=...) directly; in daemon mode the adapter wraps the same SemanticMemory instance from AgentContextCache.

1. Custom Toolkit

Create a Toolkit subclass with methods decorated by @tool(). JSON Schema is auto-extracted from type hints and docstrings.

from hive import Toolkit, tool

class WeatherToolkit(Toolkit):
    """Tools for checking weather conditions."""

    @tool()
    async def get_weather(self, city: str, units: str = "celsius") -> str:
        """Get current weather for a city.

        Args:
            city: City name
            units: Temperature units (celsius or fahrenheit)
        """
        return f"Weather in {city}: 22°{units[0].upper()}, sunny"

    @tool()
    def list_cities(self) -> str:
        """List available cities."""
        return "London, Tokyo, New York"

# Test
def test_weather_toolkit():
    tk = WeatherToolkit()
    tools = tk.get_tools()
    assert any(t.name == "get_weather" for t in tools)
    schema = tools[0].to_schema()
    assert "city" in schema["function"]["parameters"]["properties"]

2. Custom Model Provider

Subclass BaseProvider and implement the three generation methods. Add routing in factory.py.

from hive import BaseProvider, GenerateResult, Message

class MyProvider(BaseProvider):
    @property
    def available(self) -> bool:
        return True

    async def generate_with_metadata(
        self,
        messages: list[Message],
        tools: list[dict] | None = None,
        temperature: float = 0.0,
        max_tokens: int = 4096,
    ) -> GenerateResult:
        # Call your LLM here
        return GenerateResult(
            message=Message.assistant("Hello from MyProvider"),
            model="my-model",
            input_tokens=10,
            output_tokens=5,
        )

    async def generate_structured(
        self,
        messages: list[Message],
        output_type: type,
        temperature: float = 0.0,
        max_tokens: int = 4096,
    ):
        raise NotImplementedError

    @classmethod
    def lite(cls) -> "MyProvider":
        return cls(model="my-model-small")

    @classmethod
    def standard(cls) -> "MyProvider":
        return cls(model="my-model-medium")

# Test
import pytest

@pytest.mark.asyncio
async def test_my_provider():
    p = MyProvider(model="my-model")
    assert p.available
    result = await p.generate_with_metadata([Message.user("Hi")])
    assert result.message.content == "Hello from MyProvider"

3. Custom Stressor

Register new stressor types via StressorRegistry. Agents can then experience stressors beyond the 6 built-in types.

from hive import StressorRegistry, SufferingState

# Register a custom stressor
registry = StressorRegistry.default()
registry.register("burnout", escalation_rate=0.05, description="Chronic overwork exhaustion")

# Use it
state = SufferingState(agent_id="agent-1")
state.add_stressor("burnout", "Worked 50 cycles straight", "Take a rest cycle")

# Test
def test_custom_stressor():
    from hive import StressorRegistry, SufferingState

    reg = StressorRegistry.default()
    reg.register("burnout", 0.05, "Chronic overwork")

    s = SufferingState(agent_id="test")
    s.add_stressor("burnout", "overworked", "rest")
    assert len(s.active) == 1
    assert s.active[0].type == "burnout"
    assert s.active[0].escalation_per_day == 0.05

3b. Custom Mood Model

An agent's mood is derived from happiness + suffering (no stored state) and added to the goal-pursuit prompt. Swap the default circumplex model via MoodRegistry.

from hive import MoodRegistry, MoodState

class StoicMood:
    """Always calm, regardless of signals."""
    def derive(self, happiness: float, suffering_load: float, in_crisis: bool) -> MoodState:
        return MoodState("stoic", valence=0.0, arousal=0.0, note="unmoved; proceed steadily")

MoodRegistry.default().set_model(StoicMood())

# Test
def test_custom_mood_model():
    from hive import CircumplexMood, MoodRegistry, MoodState

    class StoicMood:
        def derive(self, happiness, suffering_load, in_crisis):
            return MoodState("stoic", 0.0, 0.0, "unmoved")

    reg = MoodRegistry.default()
    try:
        reg.set_model(StoicMood())
        assert reg.derive(0.1, 0.9, True).label == "stoic"
    finally:
        reg.set_model(CircumplexMood())  # restore the default (public API)

4. Custom A2A Pattern

Subclass A2APattern and register it with PatternRegistry.

from hive.interactions import A2APattern, A2AStore, A2AMessage, A2AMessageType, PatternRegistry

class BrainstormPattern(A2APattern):
    """All participants contribute ideas in parallel, then vote."""

    async def execute(
        self,
        store: A2AStore,
        initiator: str,
        participants: list[str],
        context: str,
    ) -> dict:
        message_ids = []
        for agent_id in participants:
            msg = A2AMessage(
                type=A2AMessageType.REQUEST,
                from_agent=initiator,
                to_agent=agent_id,
                subject="Brainstorm",
                body=f"Share ideas on: {context}",
                expects_reply=True,
                metadata={"brainstorm": True},
            )
            await store.send(msg)
            message_ids.append(msg.message_id)
        return {"pattern": "brainstorm", "status": "ideas_requested", "message_ids": message_ids}

# Register
PatternRegistry.default().register("brainstorm", BrainstormPattern())

# Test
import pytest
from hive.interactions import A2AStore, PatternRegistry

@pytest.mark.asyncio
async def test_brainstorm_pattern(tmp_path):
    PatternRegistry.default().register("brainstorm", BrainstormPattern())
    store = A2AStore(tmp_path)
    pattern = PatternRegistry.default().get("brainstorm")
    result = await pattern.execute(store, "agent-a", ["agent-b", "agent-c"], "new features")
    assert result["status"] == "ideas_requested"
    assert len(result["message_ids"]) == 2

5. Custom Goal Strategy

Implement the GoalStrategy protocol to control how agents generate goals when idle. Every call must return a GeneratedGoal with cost_usd and tokens set from any LLM work, even when objective is None.

from hive import GoalStrategy, GoalContext, GeneratedGoal
from pathlib import Path

class PrioritizedGoalStrategy:
    """Generate goals based on suffering and nudges."""

    async def generate_goal(self, context: GoalContext) -> GeneratedGoal:
        # Prioritize user nudges
        if context.nudges:
            return GeneratedGoal(
                objective=f"Address user request: {context.nudges[0]}",
            )
        # High suffering → self-care
        if context.suffering.cumulative_load > 0.7:
            return GeneratedGoal(
                objective="Focus on resolving active stressors",
            )
        # LLM call with no actionable goal — still report spend
        return GeneratedGoal(objective=None, cost_usd=0.0, tokens=0)

# Pass to daemon
from hive import HiveDaemon
daemon = HiveDaemon(hive_dir=Path(".hive"), goal_strategy=PrioritizedGoalStrategy())

# Test
import pytest
from hive import GoalContext, GeneratedGoal, SufferingState
from hive.agents.profile import AgentProfile

@pytest.mark.asyncio
async def test_prioritized_strategy():
    strategy = PrioritizedGoalStrategy()
    ctx = GoalContext(
        agent_id="test",
        profile=AgentProfile(name="t", role="test"),
        persona=None,
        suffering=SufferingState(agent_id="test"),
        peer_summaries=[],
        nudges=["Please write docs"],
        recent_goals=[],
    )
    result = await strategy.generate_goal(ctx)
    assert result.objective is not None
    assert "write docs" in result.objective.lower()

6. Daemon Hooks

Register callbacks for lifecycle events. Callbacks can be sync or async.

from hive import HiveDaemon, HookRegistry

daemon = HiveDaemon(hive_dir=Path(".hive"))

# Track all completed goals
completed_goals = []

async def on_goal_completed(agent_id: str, goal_id: str, **kwargs):
    completed_goals.append({"agent": agent_id, "goal": goal_id})

daemon.hooks.on("goal_completed", on_goal_completed)

# Available events:
# cycle_start(agent_id, cycle_num)
# cycle_end(agent_id, cycle_num, result)
# goal_generated(agent_id, goal_id, objective)
# goal_completed(agent_id, goal_id)
# goal_abandoned(agent_id, goal_id)
# suffering_changed(agent_id, suffering_state)

# Test
import pytest
from hive.daemon.hooks import HookRegistry

@pytest.mark.asyncio
async def test_hook_fires():
    hooks = HookRegistry()
    captured = []
    hooks.on("goal_completed", lambda **kw: captured.append(kw))
    await hooks.emit("goal_completed", agent_id="a1", goal_id="g1")
    assert captured[0] == {"agent_id": "a1", "goal_id": "g1"}

6b. Phase Guards

Veto entry into a cycle phase. The daemon divides each agent cycle into phases (approval_gate, suffering_escalation, context_assembly, goal_pursuit, goal_generation, cleanup); a guard registered for a phase can block it -- the cycle then returns "guarded" instead of running that phase. Built-ins: CostBudgetGuard (budget kill-switch on goal phases) and ManualPauseGuard (daemon freeze on all phases, wired by default).

Safety-critical guards should pass fail_closed=True to register_guard so an internal error blocks the phase instead of allowing it through (the daemon applies this to built-in budget guards when daemon.guards_fail_closed is true).

from hive.daemon.gates import CostBudgetGuard
from hive.daemon.phase import CyclePhase, PhaseGate

class NightShiftGuard:
    """Only let agents burn tokens during office hours."""

    async def should_proceed(self, gate: PhaseGate) -> bool:
        from datetime import datetime
        return 9 <= datetime.now().hour < 18

daemon.hooks.register_guard(CyclePhase.GOAL_PURSUIT, NightShiftGuard())

# Safety-critical example: block the phase if the guard raises
daemon.hooks.register_guard(
    CyclePhase.GOAL_PURSUIT, CostBudgetGuard(), fail_closed=True
)

# Test
import pytest
from hive.daemon.hooks import HookRegistry
from hive.daemon.phase import CyclePhase, PhaseGate

@pytest.mark.asyncio
async def test_guard_vetoes():
    hooks = HookRegistry()

    class Deny:
        async def should_proceed(self, gate: PhaseGate) -> bool:
            return False

    hooks.register_guard(CyclePhase.GOAL_PURSUIT, Deny())
    gate = PhaseGate(phase=CyclePhase.GOAL_PURSUIT, agent_id="a1", cycle_num=1)
    assert await hooks.check_guards(gate) is False

6c. Custom Wake Source

Wake the daemon before the heartbeat elapses. The daemon races every registered WakeSource against the heartbeat timer and starts the next cycle as soon as one fires. A2AWakeSource (A2A inbox) and NudgeWakeSource (nudge marker files) are wired by default; optional daemon.watch_files adds FileWakeSource per path. Implement async def wait(self) -> str returning a short event label:

import asyncio

class WebhookWakeSource:
    """Wake when an external system drops a marker file."""

    def __init__(self, marker: Path):
        self._marker = marker

    async def wait(self) -> str:
        while not self._marker.exists():
            await asyncio.sleep(1.0)
        self._marker.unlink()
        return "webhook"

daemon.add_wake_source(WebhookWakeSource(Path(".hive/wake.marker")))

6d. Custom Swarm Policy

Control how the daemon logs swarm-learning recommendations (e.g. role respecialization hints). DefaultSwarmPolicy emits verbose routing-hint logs; PassiveSwarmPolicy is the default (debug-level only). Neither policy mutates goals or routes tasks autonomously. Implement handle_recommendation and pass the policy to the daemon:

from hive.agents.swarm_policy import PassiveSwarmPolicy

daemon = HiveDaemon(hive_dir=Path(".hive"), swarm_policy=PassiveSwarmPolicy())

7. Custom Agent Profile

Create a YAML file in profiles/ with agent configuration. Available immediately via hive spawn <name>.

# profiles/analyst.yaml
name: analyst
role: "Analyze data, find patterns, and produce reports"
model: claude-sonnet-4-6

personality:
  traits: ["analytical", "thorough", "detail-oriented"]
  style: "Methodical and evidence-based. Always cites sources."

persona:
  values: ["accuracy", "objectivity", "clarity"]
  fears: ["drawing wrong conclusions", "missing key data"]
  purpose: "Transform raw data into actionable insights"
  long_term_goals:
    - "Build comprehensive analysis frameworks"
    - "Develop pattern recognition expertise"
  risk_tolerance: 0.2
  social_drive: 0.4

tools:
  - file_read
  - file_write
  - shell_exec
  - web_search

workspace: ./workspaces/analyst
autonomy: high
max_steps: 30

system_prompt: |
  You are a data analyst working in an isolated workspace.
  Focus on producing clear, accurate analyses.

# Test
from hive.agents.profile import AgentProfile

def test_profile_loads():
    profile = AgentProfile.from_yaml("profiles/analyst.yaml")
    assert profile.name == "analyst"
    assert "file_read" in profile.tools

8. Custom STT Provider

Implement the STTProvider protocol to add a new speech-to-text backend.

from pathlib import Path
from hive.stt.base import STTProvider, TranscriptionResult

class MySTTProvider:
    """Custom speech-to-text provider."""

    @property
    def available(self) -> bool:
        return True  # Check if your backend is configured

    async def transcribe(self, audio_path: Path) -> TranscriptionResult:
        text = await my_backend_transcribe(str(audio_path))
        return TranscriptionResult(
            text=text,
            language="en",
            duration_ms=0,
            provider="my-provider",
        )

    async def transcribe_bytes(
        self, audio: bytes, sample_rate: int = 16000
    ) -> TranscriptionResult:
        text = await my_backend_transcribe_bytes(audio)
        return TranscriptionResult(text=text, provider="my-provider")

# Test
import pytest
from hive.stt.base import STTProvider

def test_custom_stt():
    provider = MySTTProvider()
    assert isinstance(provider, STTProvider)
    assert provider.available

Built-in providers: WhisperLocal (mlx-whisper / faster-whisper), GroqSTT, DeepgramSTT. Use create_stt_provider() for auto-detection.

9. Custom Intent Router

Use IntentRouter for LLM-based text classification with user-defined intents.

from hive.routing import IntentRouter
from hive.models.groq import Groq

router = IntentRouter(
    model=Groq.lite(),
    intents={
        "task": "user wants to create a todo item",
        "note": "user wants to save information",
        "query": "user is asking a question",
    },
    fallback="query",  # default if classification fails
)

result = await router.classify("remind me to buy milk")
print(result.intent, result.confidence)  # "task", 0.95

# Test
from unittest.mock import AsyncMock, MagicMock
from hive.routing import IntentRouter
from hive.runtime.types import Message

def test_intent_router():
    provider = MagicMock()
    provider.generate = AsyncMock(
        return_value=Message.assistant('{"intent": "task", "confidence": 0.9}')
    )
    router = IntentRouter(model=provider, intents={"task": "todos", "note": "notes"})
    # router.classify() returns IntentResult with intent, confidence, raw_text

10. Custom Trigger

Register callbacks that fire on external events -- hotkeys or HTTP webhooks.

from hive.triggers import HotkeyTrigger, WebhookTrigger

# Hotkey (requires: pip install hive-agent[hotkeys])
hotkey = HotkeyTrigger()
hotkey.register("cmd+shift+m", my_callback, name="mic-toggle")
hotkey.start()  # listens in background thread

# Webhook (stdlib asyncio, no extra deps)
webhook = WebhookTrigger(host="127.0.0.1", port=8421)
webhook.register("/trigger/process", my_handler, method="POST")
await webhook.start()

# Test
import pytest
from hive.triggers import WebhookTrigger

@pytest.mark.asyncio
async def test_webhook():
    received = []
    wh = WebhookTrigger(port=0)  # ephemeral port
    wh.register("/test", lambda body: received.append(body))
    await wh.start()
    # ... send HTTP request ...
    await wh.stop()
    assert len(received) == 1

Implement the Trigger protocol for custom trigger types.

11. Plugin System

Drop a Python file containing a Toolkit subclass in .hive/plugins/. Enable plugin loading in .hive/config.yaml first (disabled by default for security -- plugins execute arbitrary Python with full process privileges, and agents can write new plugin files that hot-load every 10 daemon cycles):

plugins:
  enabled: true
  allowlist: []   # optional: restrict to named files/stems

Once enabled, plugins are auto-discovered and loaded every 10 daemon cycles.

# .hive/plugins/calculator.py
from hive import Toolkit, tool

class CalculatorToolkit(Toolkit):
    """Math tools for agents."""

    @tool()
    def add(self, a: float, b: float) -> str:
        """Add two numbers."""
        return str(a + b)

    @tool()
    def multiply(self, a: float, b: float) -> str:
        """Multiply two numbers."""
        return str(a * b)

No registration needed -- the plugin loader discovers Toolkit subclasses automatically. Tools appear in every agent's toolkit on the next plugin reload cycle.

# Test
def test_plugin_discovered():
    from hive.runtime.plugin_loader import PluginLoader
    loader = PluginLoader([Path(".hive/plugins")])
    toolkits = loader.load()
    names = [tk.__name__ for tk in toolkits]
    assert "CalculatorToolkit" in names

12. Custom World Content (Events & Jobs)

The life-event and job catalogs are registry-driven (mirroring StressorRegistry). Register your own without editing the catalog modules, or pass a custom registry to an EventEngine / WorldState for an isolated content set.

from hive.world.registry import EventRegistry, JobRegistry
from hive.world.events import LifeEvent, Choice, StatEffect
from hive.world.state import Job

# Add a new life event to the default catalog
EventRegistry.default().register(
    LifeEvent(
        event_id="found_wallet",
        name="Found a Wallet",
        description="You found a wallet on the street.",
        category="luck",
        choices=[
            Choice(id="keep", description="Keep it", stat_effects=[StatEffect(stat="money", change=80)]),
            Choice(id="return", description="Return it", stat_effects=[StatEffect(stat="happiness", change=8, change_type="percent")]),
        ],
    )
)

# Add a new job
JobRegistry.default().register(Job(job_id="pilot", title="Pilot", salary=180.0, required_skills=["flying"]))

# Test
def test_custom_event_registered():
    from hive.world.registry import EventRegistry
    from hive.world.events import LifeEvent

    EventRegistry._reset()
    reg = EventRegistry.default()
    reg.register(LifeEvent(event_id="lucky", name="Lucky", description="!", category="luck", choices=[]))
    assert reg.get("lucky") is not None

An EventEngine(stats, world, events=my_registry) fires only the events in my_registry; a WorldState seeds its jobs from JobRegistry.default() at construction, so register custom jobs before creating it.

9. Gate a Tool Behind Human Approval

Mark a tool so it pauses for human approval before running. The agent must be constructed with an ApprovalGate (the daemon wires one when approval.enabled is set in config) for the flag to take effect.

from hive import Toolkit, tool

class DeployToolkit(Toolkit):
    @tool(requires_approval=True)
    def deploy(self, target: str) -> str:
        """Deploy to an environment.

        Args:
            target: Environment name (e.g. "prod").
        """
        return f"deployed to {target}"

Custom gate via the ApprovalGate protocol:

from hive.runtime.approval import ApprovalGate, ApprovalResult, ApprovalDecision
from hive.tools.base import Tool

class AllowlistGate:
    def requires_approval(self, tool: Tool) -> bool:
        return tool.name == "deploy"

    async def check(self, tool_name: str, arguments: dict) -> ApprovalResult:
        if arguments.get("target") == "staging":
            return ApprovalResult(ApprovalDecision.APPROVED, "auto")
        return ApprovalResult(ApprovalDecision.PENDING, "needs-review")

agent = Agent(name="releaser", model=provider, toolkits=[DeployToolkit()],
              approval_gate=AllowlistGate())

The built-in StoreApprovalGate persists pending approvals so they survive the daemon's heartbeat cycles. See Human-in-the-Loop Approvals.

10. Custom Guardrail

A guardrail inspects text entering or leaving the model. Implement the Guardrail protocol and register it so it composes with the built-in PII and prompt-injection guardrails.

from hive.runtime.guardrails import (
    Guardrail, GuardrailAction, GuardrailFinding, GuardrailStage, GuardrailRegistry,
)

class ProfanityGuardrail:
    name = "profanity"
    action = GuardrailAction.REDACT

    def inspect(self, text: str, stage: GuardrailStage) -> GuardrailFinding:
        if stage is GuardrailStage.OUTPUT and "badword" in text.lower():
            cleaned = text.replace("badword", "[REDACTED]")
            return GuardrailFinding(True, self.action, cleaned, ["profanity"])
        return GuardrailFinding(False, self.action, text)

# Compose directly into a pipeline:
from hive.runtime.guardrails import GuardrailPipeline, PIIGuardrail
pipeline = GuardrailPipeline([ProfanityGuardrail(), PIIGuardrail()])
agent = Agent(name="writer", model=provider, guardrails=pipeline)

See Guardrails.