HAP Protocol¶

The Haive Agent Protocol (HAP) provides a JSON-RPC 2.0 based protocol for agent communication and orchestration.

Overview¶

HAP enables:

Agent Discovery: Find and list available agents
Remote Execution: Execute agents over network
Resource Management: Access shared resources
Progress Tracking: Real-time execution updates
Authentication: Secure agent access

Protocol Structure¶

HAP follows JSON-RPC 2.0 with extensions:

Request Format¶

{
    "jsonrpc": "2.0",
    "method": "agent/execute",
    "params": {
        "agent": "analyzer",
        "input": {
            "text": "Data to analyze"
        }
    },
    "id": "req-123"
}

Response Format¶

{
    "jsonrpc": "2.0",
    "result": {
        "output": "Analysis results",
        "metadata": {
            "duration": 1.5,
            "tokens": 150
        }
    },
    "id": "req-123"
}

HAPContext¶

The execution context for HAP operations.

Context Features¶

from haive.hap.hap import HAPContext

# Create context
context = HAPContext(request_id="req-123")

# Logging
await context.debug("Starting process")
await context.info("Processing data")
await context.warning("Low memory")
await context.error("Failed to connect")

# Progress tracking
await context.report_progress(0, 100, "Initializing")
await context.report_progress(50, 100, "Processing")
await context.report_progress(100, 100, "Complete")

# Data storage
context.set_data("user", "alice")
user = context.get_data("user")

Resource Providers¶

HAP supports pluggable resource providers:

SimpleResourceProvider¶

from haive.hap.hap import SimpleResourceProvider

provider = SimpleResourceProvider()

# Add resources
provider.add_resource("config://app", {"timeout": 30})
provider.add_resource("data://users", ["alice", "bob"])

# Read resources
config = await provider.read_resource("config://app")
users = await provider.read_resource("data://users")

Custom Providers¶

class DatabaseResourceProvider:
    async def read_resource(self, uri: str) -> Any:
        if uri.startswith("db://"):
            table = uri[5:]
            return await fetch_from_database(table)
        raise KeyError(f"Unknown resource: {uri}")

context = HAPContext(
    resource_provider=DatabaseResourceProvider()
)

Authentication¶

HAP includes authentication support:

SimpleAuthProvider¶

from haive.hap.hap import SimpleAuthProvider

# Create auth provider
auth = SimpleAuthProvider(
    user="alice",
    scopes=["execute", "read", "write"]
)

# Use in context
context = HAPContext(auth_provider=auth)

# Check permissions
if context.has_scope("execute"):
    # Allowed to execute agents
    result = await execute_agent()

# Get user info
user = context.get_auth_user()  # "alice"

Protocol Methods¶

Standard Methods¶

Method	Description	Parameters
`agent/list`	List available agents	`filter`, `limit`
`agent/execute`	Execute an agent	`agent`, `input`
`agent/describe`	Get agent metadata	`agent`
`resource/read`	Read a resource	`uri`
`resource/list`	List resources	`prefix`

Notifications¶

Notification	Description	Data
`progress`	Progress update	`current`, `total`, `message`
`log`	Log message	`level`, `message`, `timestamp`
`state`	State change	`old_state`, `new_state`

Implementation Example¶

Basic HAP Server¶

from haive.hap.hap import HAPServer

# Create server
server = HAPServer()

# Register agents
server.register_agent("analyzer", analyzer_agent)
server.register_agent("summarizer", summarizer_agent)

# Add resources
server.add_resource("config://agents", agent_config)

# Start server
await server.start(port=8080)

HAP Client¶

from haive.hap.hap import HAPClient

# Connect to server
client = HAPClient("http://localhost:8080")

# List agents
agents = await client.list_agents()

# Execute agent
result = await client.execute_agent(
    "analyzer",
    {"text": "Data to analyze"}
)

# Read resource
config = await client.read_resource("config://agents")

Integration with AGP¶

HAP provides the protocol layer for AGP:

# Future implementation
from haive.hap.server.runtime import HAPRuntime
from haive.hap.hap import HAPServer

# Create HAP runtime
runtime = HAPRuntime(graph)

# Expose via HAP
server = HAPServer()
server.expose_runtime(runtime)

# Agents available via protocol
await server.start()

Security Considerations¶

Authentication¶

Use OAuth2/JWT tokens for production
Implement scope-based permissions
Rotate credentials regularly

Transport Security¶

Always use HTTPS in production
Implement request signing
Validate all inputs

Rate Limiting¶

Implement per-user rate limits
Use exponential backoff
Monitor for abuse

Future Development¶

Full Protocol Implementation: Complete JSON-RPC server
MCP Bridge: Interoperability with MCP
Discovery Service: Agent registry and discovery
Streaming Support: Real-time data streams
Federation: Multi-server agent networks