haive-hap

Haive Agent Protocol — "MCP for Agents"

A protocol and runtime for orchestrating multiple AI agents in complex workflows. Where MCP standardizes tool integration, HAP standardizes agent integration. Define workflows as graphs, run them locally or as JSON-RPC services, and compose specialized agents into pipelines that solve problems no single agent can.

---

Why HAP?

You have specialized agents — researcher, writer, fact-checker, analyzer. Each is good at one thing. To solve real problems, you need to coordinate them: pass output from one to the next, run some in parallel, route based on conditions, handle errors gracefully.

You could hand-roll this for every project. Or you could use HAP.

HAP gives you:

• Workflow-as-graph — define agent pipelines as HAPGraph objects
• Three execution modes — Sequential, Parallel, Conditional
• Local + Remote — In-process for simple cases, JSON-RPC server for distributed
• Dynamic loading — Load agents via entrypoint strings (haive.agents.simple.agent:SimpleAgent)
• Lifecycle hooks — Pre/post execution, error recovery, retry logic
• Execution tracking — Path traversed, timing per node, agent metadata
• Type safety — Pydantic models throughout, validated state transitions

---

Installation

pip install haive-hap

---

Core Concepts

HAPGraph

Defines an agent workflow as a directed graph. Each node is an agent (or agent reference), each edge is a transition.

from haive.hap.models import HAPGraph
from haive.agents.simple.agent import SimpleAgent
from haive.core.engine.aug_llm import AugLLMConfig

researcher = SimpleAgent(name="researcher", engine=AugLLMConfig())
analyzer = SimpleAgent(name="analyzer", engine=AugLLMConfig(temperature=0.3))
writer = SimpleAgent(name="writer", engine=AugLLMConfig(temperature=0.8))

graph = HAPGraph()
graph.add_agent_node("research", researcher, next_nodes=["analyze"])
graph.add_agent_node("analyze", analyzer, next_nodes=["write"])
graph.add_agent_node("write", writer)
graph.entry_node = "research"

HAPRuntime

Executes a graph. Handles state transitions, error recovery, and observability.

from haive.hap.server.runtime import HAPRuntime

runtime = HAPRuntime(graph)
result = await runtime.run({
    "topic": "AI safety in 2025",
    "depth": "comprehensive",
})

print(f"Execution path: {result.execution_path}")
print(f"Final outputs: {result.outputs}")
print(f"Per-node timing: {result.timing}")

HAPContext

State that flows through the workflow. Each node can read and contribute to the context.

result = await runtime.run({"input": "..."})
# context flows: research -> analyze -> write
# each node sees the previous nodes' outputs

---

Workflow Patterns

1. Sequential — Output Chain

Each agent processes the output of the previous agent.

graph = HAPGraph()
graph.add_agent_node("step1", a1, next_nodes=["step2"])
graph.add_agent_node("step2", a2, next_nodes=["step3"])
graph.add_agent_node("step3", a3)
graph.entry_node = "step1"

Use cases: Research pipelines, content workflows (research → write → review), data processing chains.

2. Parallel Fork-Join

A coordinator dispatches to N workers, results are combined.

graph = HAPGraph()
graph.add_agent_node("coordinator", coord, next_nodes=["w1", "w2", "w3"])
graph.add_agent_node("w1", worker1, next_nodes=["combine"])
graph.add_agent_node("w2", worker2, next_nodes=["combine"])
graph.add_agent_node("w3", worker3, next_nodes=["combine"])
graph.add_agent_node("combine", combiner)
graph.entry_node = "coordinator"

Use cases: Multi-perspective analysis, parallel data processing, ensemble methods.

3. Conditional Routing

Route based on the output of a classifier.

graph = HAPGraph()
graph.add_agent_node("classifier", classifier, next_nodes=["technical", "creative", "general"])
graph.add_agent_node("technical", tech_agent)
graph.add_agent_node("creative", creative_agent)
graph.add_agent_node("general", general_agent)
graph.entry_node = "classifier"

Use cases: Customer service routing, content classification → specialized handler, multi-modal processing.

4. Error Recovery

Workflow with retry and fallback.

graph = HAPGraph()
graph.add_agent_node("primary", primary_agent, next_nodes=["validator"])
graph.add_agent_node("validator", validator, next_nodes=["fallback", "success"])
graph.add_agent_node("fallback", fallback_agent, next_nodes=["success"])
graph.add_agent_node("success", final_agent)

---

Local vs Remote Execution

Local (In-Process)

Fast, no network. Perfect for development and most production use cases.

runtime = HAPRuntime(graph)
result = await runtime.run(input_data)

Remote (JSON-RPC Server)

Run workflows as a service. Multiple clients can submit jobs to the same workflow.

# Server side
from haive.hap.hap.server import HAPServer

server = HAPServer(workflow_graph=my_graph)
await server.start(host="localhost", port=8080)

# Client side
from haive.hap.client.remote import RemoteClient

client = RemoteClient("http://localhost:8080")
result = await client.execute_workflow({
    "workflow_id": "research_pipeline",
    "input": {"topic": "Climate change solutions"}
})

---

Dynamic Agent Loading

Load agents at runtime via entrypoint strings — perfect for distributed deployments where agents are defined elsewhere.

graph = HAPGraph()
graph.add_entrypoint_node(
    "analyzer",
    "haive.agents.simple.agent:SimpleAgent",
    next_nodes=["formatter"]
)
graph.add_entrypoint_node(
    "formatter",
    "haive.agents.simple.agent:SimpleAgent"
)
graph.entry_node = "analyzer"

# Runtime loads agents on demand
runtime = HAPRuntime(graph)

---

Documentation

📖 Full documentation: https://pr1m8.github.io/haive-hap/

---

Related Packages

Package	Description
haive-core	Foundation: engines, graphs
haive-agents	Production agents (used in HAP workflows)
haive-mcp	MCP integration

---

License

MIT © pr1m8