file_templates

Agent File Templates

Complete code templates for each file in a Hive agent package.

config.py

"""Runtime configuration."""

import json
from dataclasses import dataclass, field
from pathlib import Path


def _load_preferred_model() -> str:
    """Load preferred model from ~/.hive/configuration.json."""
    config_path = Path.home() / ".hive" / "configuration.json"
    if config_path.exists():
        try:
            with open(config_path) as f:
                config = json.load(f)
            llm = config.get("llm", {})
            if llm.get("provider") and llm.get("model"):
                return f"{llm['provider']}/{llm['model']}"
        except Exception:
            pass
    return "anthropic/claude-sonnet-4-20250514"


@dataclass
class RuntimeConfig:
    model: str = field(default_factory=_load_preferred_model)
    temperature: float = 0.7
    max_tokens: int = 40000
    api_key: str | None = None
    api_base: str | None = None


default_config = RuntimeConfig()


@dataclass
class AgentMetadata:
    name: str = "My Agent Name"
    version: str = "1.0.0"
    description: str = "What this agent does."
    intro_message: str = "Welcome! What would you like me to do?"


metadata = AgentMetadata()

nodes/__init__.py

"""Node definitions for My Agent."""

from framework.orchestrator import NodeSpec

# Node 1: Process (autonomous entry node)
# The queen handles intake and passes structured input via
# run_agent_with_input(task). NO client-facing intake node.
# The queen defines input_keys at build time and fills them at run time.
process_node = NodeSpec(
    id="process",
    name="Process",
    description="Execute the task using available tools",
    node_type="event_loop",
    max_node_visits=0,  # Unlimited for forever-alive
    input_keys=["user_request", "feedback"],
    output_keys=["results"],
    nullable_output_keys=["feedback"],  # Only on feedback edge
    success_criteria="Results are complete and accurate.",
    system_prompt="""\
You are a processing agent. Your task is in memory under "user_request". \
If "feedback" is present, this is a revision — address the feedback.

Work in phases:
1. Use tools to gather/process data
2. Analyze results
3. Call set_output in a SEPARATE turn:
   - set_output("results", "structured results")
""",
    tools=["web_search", "web_scrape", "save_data", "load_data", "list_data_files"],
)

# Node 2: Handoff (autonomous)
handoff_node = NodeSpec(
    id="handoff",
    name="Handoff",
    description="Prepare worker results for queen review",
    node_type="event_loop",
    client_facing=False,
    max_node_visits=0,
    input_keys=["results", "user_request"],
    output_keys=["next_action", "feedback", "worker_summary"],
    nullable_output_keys=["feedback", "worker_summary"],
    success_criteria="Results are packaged for queen decision-making.",
    system_prompt="""\
Do NOT talk to the user directly. The queen is the only user interface.

If blocked by tool failures, missing credentials, or unclear constraints, call:
- escalate(reason, context)
Then set:
- set_output("next_action", "escalated")
- set_output("feedback", "what help is needed")

Otherwise summarize findings for queen and set:
- set_output("worker_summary", "short summary for queen")
- set_output("next_action", "done") or set_output("next_action", "revise")
- set_output("feedback", "what to revise") only when revising
""",
    tools=[],
)

__all__ = ["process_node", "handoff_node"]

agent.py

"""Agent graph construction for My Agent."""

from pathlib import Path

from framework.orchestrator import EdgeSpec, EdgeCondition, Goal, SuccessCriterion, Constraint
from framework.orchestrator.edge import GraphSpec
from framework.orchestrator.orchestrator import ExecutionResult
from framework.orchestrator.checkpoint_config import CheckpointConfig
from framework.llm import LiteLLMProvider
from framework.loader.tool_registry import ToolRegistry
from framework.host.agent_host import AgentHost
from framework.host.execution_manager import EntryPointSpec


from .config import default_config, metadata
from .nodes import process_node, handoff_node

# Goal definition
goal = Goal(
    id="my-agent-goal",
    name="My Agent Goal",
    description="What this agent achieves.",
    success_criteria=[
        SuccessCriterion(id="sc-1", description="...", metric="...", target="...", weight=0.5),
        SuccessCriterion(id="sc-2", description="...", metric="...", target="...", weight=0.5),
    ],
    constraints=[
        Constraint(id="c-1", description="...", constraint_type="hard", category="quality"),
    ],
)

# Node list
nodes = [process_node, handoff_node]

# Edge definitions
edges = [
    EdgeSpec(id="process-to-handoff", source="process", target="handoff",
             condition=EdgeCondition.ON_SUCCESS, priority=1),
    # Feedback loop — revise results
    EdgeSpec(id="handoff-to-process", source="handoff", target="process",
             condition=EdgeCondition.CONDITIONAL,
             condition_expr="str(next_action).lower() == 'revise'", priority=2),
    # Escalation loop — queen injects guidance and worker retries
    EdgeSpec(id="handoff-escalated", source="handoff", target="process",
             condition=EdgeCondition.CONDITIONAL,
             condition_expr="str(next_action).lower() == 'escalated'", priority=3),
    # Loop back for next task after queen decision
    EdgeSpec(id="handoff-done", source="handoff", target="process",
             condition=EdgeCondition.CONDITIONAL,
             condition_expr="str(next_action).lower() == 'done'", priority=1),
]

# Graph configuration — entry is the autonomous process node
# The queen handles intake and passes the task via run_agent_with_input(task)
entry_node = "process"
entry_points = {"start": "process"}
pause_nodes = []
terminal_nodes = []  # Forever-alive

# Module-level vars read by AgentRunner.load()
conversation_mode = "continuous"
identity_prompt = "You are a helpful agent."
loo