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.
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
HAPGraphobjects - 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
pip install haive-hapDefines an agent workflow as a directed graph. Each node is an agent (or agent reference), each edge is a transition.
fromhaive.hap.modelsimportHAPGraphfromhaive.agents.simple.agentimportSimpleAgentfromhaive.core.engine.aug_llmimportAugLLMConfigresearcher=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"Executes a graph. Handles state transitions, error recovery, and observability.
fromhaive.hap.server.runtimeimportHAPRuntimeruntime=HAPRuntime(graph)
result=awaitruntime.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}")State that flows through the workflow. Each node can read and contribute to the context.
result=awaitruntime.run({"input": "..."})
# context flows: research -> analyze -> write# each node sees the previous nodes' outputsEach 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.
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.
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.
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)Fast, no network. Perfect for development and most production use cases.
runtime=HAPRuntime(graph)
result=awaitruntime.run(input_data)Run workflows as a service. Multiple clients can submit jobs to the same workflow.
# Server sidefromhaive.hap.hap.serverimportHAPServerserver=HAPServer(workflow_graph=my_graph)
awaitserver.start(host="localhost", port=8080)
# Client sidefromhaive.hap.client.remoteimportRemoteClientclient=RemoteClient("http://localhost:8080")
result=awaitclient.execute_workflow({
"workflow_id": "research_pipeline",
"input": {"topic": "Climate change solutions"}
})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 demandruntime=HAPRuntime(graph)📖 Full documentation:https://pr1m8.github.io/haive-hap/
| Package | Description |
|---|---|
| haive-core | Foundation: engines, graphs |
| haive-agents | Production agents (used in HAP workflows) |
| haive-mcp | MCP integration |
MIT © pr1m8