haive-hap 1.0.0

Haive Agent Protocol - MCP for Agents

HAP - Haive Agent Protocol

"MCP for Agents" - A protocol for orchestrating multiple AI agents in complex workflows.

🎯 Overview

HAP (Haive Agent Protocol) enables seamless orchestration of multiple AI agents, similar to how MCP (Model Context Protocol) exposes tools and resources. It provides a standardized way to coordinate agents working together on complex multi-step problems that no single agent can solve alone.

Key Features

• 🔄 Multi-Agent Orchestration - Sequential, parallel, and conditional workflows
• 🌐 Protocol-Based Communication - JSON-RPC 2.0 standard for remote agents
• 📊 State Management - Context flows seamlessly between agents
• 🛠️ Tool Integration - Access individual agent tools directly
• 📈 Performance Monitoring - Track execution times and statistics
• 🔌 Flexible Deployment - Local in-process or distributed network execution

🚀 Quick Start

Basic Agent Workflow

from haive.hap.models.graph import HAPGraph
from haive.hap.server.runtime import HAPRuntime
from haive.agents.simple import SimpleAgent
from haive.core.engine.aug_llm import AugLLMConfig

# Create specialized agents
researcher = SimpleAgent(
    name="researcher",
    engine=AugLLMConfig(
        temperature=0.7,
        system_message="You gather comprehensive information on topics."
    )
)

analyst = SimpleAgent(
    name="analyst",
    engine=AugLLMConfig(
        temperature=0.3,
        system_message="You analyze findings and identify key insights."
    )
)

# Build workflow graph
graph = HAPGraph()
graph.add_agent_node("research", researcher, next_nodes=["analyze"])
graph.add_agent_node("analyze", analyst)
graph.entry_node = "research"

# Execute workflow
runtime = HAPRuntime(graph)
result = await runtime.run({"topic": "AI in Healthcare"})

print(f"Workflow path: {result.execution_path}")
print(f"Final output: {result.outputs}")

Parallel Agent Execution

# Create agents for parallel analysis
sentiment_agent = SimpleAgent(name="sentiment", engine=sentiment_config)
topic_agent = SimpleAgent(name="topics", engine=topic_config)
summary_agent = SimpleAgent(name="summary", engine=summary_config)

# Build parallel workflow
graph = HAPGraph()
graph.add_agent_node("sentiment", sentiment_agent, next_nodes=["combine"])
graph.add_agent_node("topics", topic_agent, next_nodes=["combine"])
graph.add_agent_node("summary", summary_agent, next_nodes=["combine"])
graph.add_agent_node("combine", combiner_agent)

# Multiple entry points for parallel execution
graph.entry_node = ["sentiment", "topics", "summary"]

runtime = HAPRuntime(graph)
result = await runtime.run({"text": "Customer feedback data..."})

Agent as Service

from haive.hap.servers.agent import AgentServer

# Wrap any agent as HAP server
server = AgentServer(
    agent=my_agent,
    expose_tools=True,      # Expose individual tools
    expose_state=True       # Expose agent state
)

# Access agent capabilities
info = server.get_resource("agent://my_agent/info")
stats = server.get_resource("agent://my_agent/stats")

# Execute agent
result = await server.execute({
    "input_data": {"query": "Process this request"},
    "timeout": 30.0
})

📦 Installation

# Install with poetry
poetry add haive-hap

# Or with pip
pip install haive-hap

🏗️ Architecture

Core Components

1. Graph Definition (models/graph.py)

   • HAPGraph - Define agent workflows as directed graphs
   • HAPNode - Individual workflow nodes (agents, tools, decisions)
   • Support for sequential, parallel, and conditional execution

2. Runtime Engine (server/runtime.py)

   • HAPRuntime - Execute workflows with state management
   • Automatic agent loading from entrypoints
   • Error handling and recovery

3. Protocol Layer (types/protocol.py)

   • JSON-RPC 2.0 compliant messaging
   • Comprehensive Pydantic validation
   • Standard error codes and handling

4. Agent Server (servers/agent.py)

   • Expose any Haive agent as HAP service
   • Tool-level granularity
   • Resource endpoints for state and config

📊 Real-World Use Cases

1. Research Pipeline

# Research → Analysis → Report Generation
graph = HAPGraph()
graph.add_agent_node("research", research_agent, next_nodes=["analyze"])
graph.add_agent_node("analyze", analysis_agent, next_nodes=["report"])
graph.add_agent_node("report", report_agent)

2. Customer Support Workflow

# Classify → Route → Respond → Follow-up
graph = HAPGraph()
graph.add_agent_node("classify", classifier, next_nodes=["technical", "billing", "general"])
graph.add_agent_node("technical", tech_agent, next_nodes=["followup"])
graph.add_agent_node("billing", billing_agent, next_nodes=["followup"])
graph.add_agent_node("general", general_agent, next_nodes=["followup"])
graph.add_agent_node("followup", followup_agent)

3. Content Creation Pipeline

# Plan → Research → Write → Edit → Publish
graph = HAPGraph()
graph.add_agent_node("plan", planner, next_nodes=["research"])
graph.add_agent_node("research", researcher, next_nodes=["write"])
graph.add_agent_node("write", writer, next_nodes=["edit"])
graph.add_agent_node("edit", editor, next_nodes=["publish"])
graph.add_agent_node("publish", publisher)

🔧 Advanced Features

Dynamic Agent Loading

# Load agents from entrypoints (for distributed deployment)
graph = HAPGraph()
graph.add_entrypoint_node(
    "analyzer",
    "mypackage.agents:AnalyzerAgent",
    config={"temperature": 0.3}
)

Conditional Routing

# Route based on agent output
def routing_function(context):
    score = context.outputs.get("confidence", 0)
    return "expert" if score < 0.7 else "finalize"

graph.add_decision_node(
    "router",
    decision_func=routing_function,
    next_nodes={"expert": "expert_agent", "finalize": "final_agent"}
)

State Persistence

# Save and restore workflow state
runtime = HAPRuntime(graph, state_file="workflow_state.json")

# Resume from checkpoint
result = await runtime.resume_from_checkpoint()

📈 Performance Monitoring

# Get execution statistics
stats = runtime.get_stats()
print(f"Total executions: {stats['execution_count']}")
print(f"Average time: {stats['average_execution_time']:.2f}s")

# Per-node performance
for node, metadata in result.node_metadata.items():
    print(f"{node}: {metadata['execution_time']:.2f}s")

🧪 Testing

# Run all tests (real agents, no mocks)
poetry run pytest

# Test specific components
poetry run pytest tests/test_graph_execution.py -v
poetry run pytest tests/test_agent_server.py -v
poetry run pytest tests/test_hap_workflow.py -v

# Integration tests
poetry run pytest tests/integration/ -v

📚 Documentation

Building Documentation

# Build enhanced Sphinx docs
cd docs
poetry run sphinx-build -b html source build

# View locally
python -m http.server 8003 --directory build

Documentation Features

• 🎨 Enhanced Furo theme with purple/violet styling
• 💡 Interactive tooltips and code examples
• 📱 Mobile-optimized responsive design
• 🔍 Advanced search with syntax highlighting
• 📖 Hierarchical API reference organization

🤝 Contributing

1. Fork the repository
2. Create your feature branch (git checkout -b feature/amazing-feature)
3. Write tests for your changes (no mocks!)
4. Ensure all tests pass (poetry run pytest)
5. Commit your changes (git commit -m 'feat: add amazing feature')
6. Push to the branch (git push origin feature/amazing-feature)
7. Open a Pull Request

📄 License

MIT License - see LICENSE for details.

🔮 Roadmap

• v0.2.0 - WebSocket transport for real-time updates
• v0.3.0 - Distributed workflow execution
• v0.4.0 - Visual workflow builder UI
• v0.5.0 - GraphQL API interface
• v1.0.0 - Production-ready with performance optimizations

🙏 Acknowledgments

HAP is inspired by the Model Context Protocol (MCP) but adapted specifically for agent orchestration. Special thanks to the Haive framework team for providing the foundation for advanced agent development.

---

HAP - Orchestrating AI agents with elegance and power. 🚀