Python API¶

The Hive facade class provides a high-level Python API for programmatic control.

Hive Facade¶

from hive import Hive
from pathlib import Path

hive = Hive(path=Path("."))  # defaults to current directory

Methods¶

Method	Parameters	Returns	Description
`init()`		`None`	Optional. Scaffold the `.hive/` directory eagerly. Idempotent -- the directory is also created lazily on first use, so calling `init()` is no longer required.
`spawn(preset, model=None)`	preset name, optional model override	`str` (agent_id)	Spawn an agent
`start(cycles, heartbeat, profiles, fresh)`	cycles (None=forever), heartbeat seconds, profile list, fresh start	`None`	Start daemon (blocks)
`stop()`		`None`	Signal daemon to stop
`status()`		`list[dict]`	Agent statuses with id, name, role, model, status, goal
`nudge(agent, message)`	agent name/ID, message text	`None`	Send direction to agent
`kill(agent)`	agent name/ID	`None`	Terminate agent
`inspect(run_id)`	run ID string	`dict \| None`	Get run summary

Example¶

from hive import Hive

hive = Hive()  # .hive/ is scaffolded lazily; calling init() is optional

hive.spawn("coder")
hive.spawn("gambler")
hive.spawn("philosopher")

# Run 50 cycles with 5-second heartbeat
hive.start(cycles=50, heartbeat=5)

Context manager¶

Hive works as a sync or async context manager and stops the daemon on exit:

with Hive() as hive:
    hive.spawn("coder")
    hive.start(cycles=50, heartbeat=5)

Use async with when you are already inside an event loop -- setup and teardown run without a thread-pool hop. (start() runs its own loop and blocks, so drive the daemon directly via HiveDaemon in async contexts; the facade's store-backed methods -- spawn, status, nudge, kill -- are loop-safe.)

async with Hive() as hive:
    hive.spawn("coder")
    print(hive.status())

Agent¶

The core class for building agents.

from hive import Agent, Task, Persona
from hive.models.anthropic import Anthropic
from hive.runtime import FileToolkit, ShellToolkit

agent = Agent(
    name="coder",
    model=Anthropic.lite(),
    persona=Persona(name="Coder", personality=["methodical"]),
    toolkits=[FileToolkit(), ShellToolkit()],
    tool_timeout=60.0,  # per-tool wall-clock limit in seconds; 0.0 (default) disables
)

tool_timeout bounds each tool call: a tool that runs longer becomes a tool-error result rather than stalling the whole cycle. It defaults to 0.0 (disabled) for a standalone Agent; the daemon sets it from DaemonConfig.tool_timeout (60s).

Pass approval_gate=... (an ApprovalGate) to require human approval before selected tools run. A gated call pauses the run, which returns a TaskResult with status=TaskStatus.WAITING_APPROVAL; resolving the approval lets a later run execute the tool. See Gate a Tool Behind Human Approval.

The instructions argument accepts a plain str or any InstructionLike object -- anything implementing build_system_prompt(toolkit_instructions=None, response_model=None). Instructions and Persona both satisfy it, and you can supply your own. The agent never mutates the object you pass (the response model is injected per call).

Standalone (no daemon)¶

The Agent is fully usable on its own -- no daemon, no .hive directory, no global state. Construct it with a name and a model and run a task:

import asyncio
from hive import Agent, Task
from hive.models.anthropic import Anthropic

agent = Agent(name="helper", model=Anthropic.lite())
result = asyncio.run(agent.run(Task(instruction="What is 2 + 2?")))
print(result.output)

Or a synchronous one-shot, with no async at all:

agent = Agent(name="q", model=Anthropic.lite())
print(agent.run_once_sync("Capital of France?"))

Add toolkits=[...] for tools, response_model=... for validated output, or on_text=... to stream. The daemon, when used, is just another consumer of this same Agent.

Key Methods¶

Method	Returns	Description
`await agent.run(task)`	`TaskResult`	Full ReAct loop with tools
`await agent.run_once(prompt)`	`str`	Single request-response
`await agent.run_once_structured(prompt, output_type)`	`T`	Single response as Pydantic model
`agent.run_once_sync(prompt)`	`str`	Synchronous one-shot
`agent.run_once_structured_sync(prompt, output_type)`	`T`	Synchronous structured output

Streaming text¶

Pass an on_text callback to stream assistant text as it is generated. When the provider supports Capability.STREAMING the callback receives token deltas; with a non-streaming provider it receives the full text in one call (via the base fallback). The ReAct loop is otherwise unchanged -- tool calls and control flow run as usual.

agent = Agent(name="coder", model=Anthropic.standard(), on_text=lambda t: print(t, end=""))
await agent.run(Task(instruction="Explain the plan, then list the files."))

TaskResult¶

result = await agent.run(Task(instruction="Write a prime checker"))
print(result.output)       # Final response text
print(result.status)       # TaskStatus.COMPLETED | .FAILED | .MAX_STEPS
print(result.steps)        # List of steps taken
print(result.input_tokens) # Total input tokens
print(result.output_tokens)# Total output tokens

Workflow¶

Chain agents in a pipeline.

from hive.runtime import Workflow, Step

workflow = Workflow(
    name="research-pipeline",
    steps=[
        Step(name="research", agent=researcher, instruction="Research {topic}", output_key="research"),
        Step(name="draft", agent=writer, instruction="Write about: {research}", output_key="draft"),
        Step(name="review", agent=reviewer, instruction="Review: {draft}", output_key="review"),
    ],
)

result = await workflow.run({"topic": "quantum computing"})
print(result["review"])

Steps pass context between them -- {key} placeholders are replaced with outputs from previous steps.

Model Providers¶

All providers follow the same tier pattern:

from hive.models.anthropic import Anthropic
from hive.models.openai import OpenAI
from hive.models.groq import Groq
from hive.models.fireworks import Fireworks
from hive.models.ollama import Ollama
from hive.models.lmstudio import LMStudio

# Tier presets
model = Anthropic.lite()      # Haiku
model = Anthropic.standard()  # Sonnet
model = Anthropic.pro()       # Opus

# Direct construction
model = Anthropic(model="claude-sonnet-4-6")

BaseProvider Interface¶

All providers implement:

Method	Returns	Description
`await generate_with_metadata(messages, tools, temperature, max_tokens)`	`GenerateResult`	Generate with full metadata
`generate_stream(messages, tools, temperature, max_tokens)`	`AsyncIterator[StreamEvent]`	Stream `TEXT` deltas then a terminal `DONE` event
`await generate_structured(messages, output_type, temperature, max_tokens)`	`T`	Generate as Pydantic model
`available` (property)	`bool`	Whether the provider can be used
`supports(capability)`	`bool`	Whether an optional `Capability` is supported
`availability()`	`Availability`	Why the provider is or isn't usable

Capabilities and Availability¶

Branch on what a provider can do with supports() rather than special-casing provider classes, and use availability() to tell why a provider is unusable -- a missing API key reads differently from an unreachable local server.

from hive.models.anthropic import Anthropic
from hive.models.base import Availability, Capability

model = Anthropic.standard()

model.supports(Capability.TOOLS)              # True
model.supports(Capability.STRUCTURED_OUTPUT)  # True
model.supports(Capability.STREAMING)          # True for Anthropic + OpenAI-compatible

status = model.availability()
if status is not Availability.AVAILABLE:
    print(f"unusable: {status.value}")  # e.g. "no_api_key" or "unreachable"

Capability members: TOOLS, STRUCTURED_OUTPUT, STREAMING. Availability members: AVAILABLE, NO_API_KEY, UNREACHABLE, UNKNOWN.

Factory¶

from hive import create_runtime_provider

model = create_runtime_provider("anthropic:lite")
model = create_runtime_provider("openai:standard")
model = create_runtime_provider("ollama:standard")

Errors¶

Hive raises a small typed hierarchy rooted at HiveError, so you can catch the whole family at once:

from hive import AgentNotFoundError, HiveError, ProfileNotFoundError

try:
    hive.kill("ghost")
except HiveError as e:
    print(f"hive failed: {e}")

Error	Raised when	Also subclasses
`HiveError`	base for all Hive errors	`Exception`
`AgentNotFoundError`	an agent id/name/prefix can't be resolved	`ValueError`
`ProfileNotFoundError`	a preset/profile file is missing	`FileNotFoundError`

Each subclasses the builtin it replaced, so existing except ValueError / except FileNotFoundError handlers keep working.