hip-0009

HIP-9: Agent SDK - Multi-Agent Orchestration Framework. Status Draft. Hanzo's own standard — read this before implementing against it.

HIP-0009: Agent SDK - Multi-Agent Orchestration Framework

Abstract

This proposal defines the Agent SDK specification, Hanzo's multi-agent systems framework with orchestration, network routing, shared state management, and MCP support. The SDK enables building, deploying, and managing autonomous AI agents that can collaborate to solve complex tasks.

Repository: github.com/hanzoai/agent PyPI: hanzoai-agent NPM: @hanzoai/agent

Specification

Agent Architecture

class HanzoAgent:
    """
    Base agent class with full capabilities
    """
    def __init__(self, config):
        self.id = generate_agent_id()
        self.model = config.model  # LLM backend
        self.tools = config.tools  # Available tools
        self.memory = AgentMemory()  # Long-term memory
        self.state = AgentState()  # Current state
        self.network = AgentNetwork()  # P2P networking
        
    async def think(self, observation):
        """Reasoning and planning"""
        thought = await self.model.reason(
            observation=observation,
            memory=self.memory.recall(),
            tools=self.tools.available()
        )
        return thought
        
    async def act(self, thought):
        """Execute actions based on thought"""
        if thought.requires_tool:
            result = await self.tools.execute(thought.tool_call)
        elif thought.requires_collaboration:
            result = await self.network.request_help(thought.task)
        else:
            result = await self.model.generate(thought.prompt)
        return result
        
    async def observe(self, environment):
        """Perceive environment changes"""
        return await environment.get_observation(self.id)
        
    async def run(self):
        """Main agent loop"""
        while not self.state.is_complete:
            observation = await self.observe(self.environment)
            thought = await self.think(observation)
            action = await self.act(thought)
            self.state.update(action)
            self.memory.store(observation, thought, action)

Multi-Agent Orchestration

class AgentOrchestrator:
    """
    Orchestrates multiple agents for complex tasks
    """
    def __init__(self):
        self.agents = {}
        self.task_queue = PriorityQueue()
        self.shared_state = SharedState()
        
    def spawn_agent(self, role, config):
        """Create specialized agent"""
        agent = HanzoAgent(config)
        agent.role = role
        self.agents[agent.id] = agent
        return agent
        
    async def delegate_task(self, task):
        """Intelligent task delegation"""
        # Analyze task requirements
        requirements = self.analyze_task(task)
        
        # Find or create suitable agents
        if requirements.needs_research:
            researcher = self.spawn_agent("researcher", ResearchConfig())
            
        if requirements.needs_coding:
            coder = self.spawn_agent("coder", CodingConfig())
            
        if requirements.needs_review:
            reviewer = self.spawn_agent("reviewer", ReviewConfig())
            
        # Create execution plan
        plan = self.create_plan(task, self.agents)
        
        # Execute plan with coordination
        return await self.execute_plan(plan)

Agent Roles and Specializations

Built-in Roles:
  Researcher:
    model: gpt-4
    tools: [search, browse, summarize]
    skills: [information_gathering, fact_checking]
    
  Coder:
    model: claude-3-opus
    tools: [code_interpreter, debugger, linter]
    skills: [code_generation, refactoring, testing]
    
  Analyst:
    model: gemini-pro
    tools: [calculator, data_viz, statistics]
    skills: [data_analysis, visualization, reporting]
    
  Designer:
    model: dall-e-3
    tools: [image_gen, edit, style_transfer]
    skills: [visual_design, ui_ux, branding]
    
  Manager:
    model: gpt-4
    tools: [task_tracker, calendar, communication]
    skills: [planning, delegation, coordination]

Agent Network Protocol

class AgentNetwork:
    """
    P2P network for agent communication
    """
    def __init__(self):
        self.peers = {}
        self.discovery = ServiceDiscovery()
        self.router = MessageRouter()
        
    async def discover_agents(self, capability):
        """Find agents with specific capabilities"""
        return await self.discovery.find(
            capability=capability,
            max_distance=3,  # Network hops
            timeout=5000
        )
        
    async def send_message(self, recipient_id, message):
        """Direct agent-to-agent communication"""
        route = await self.router.find_route(recipient_id)
        return await self.send_via_route(route, message)
        
    async def broadcast(self, message, scope="local"):
        """Broadcast to multiple agents"""
        if scope == "local":
            recipients = self.peers.values()
        elif scope == "global":
            recipients = await self.discovery.all_agents()
            
        tasks = [self.send_message(r.id, message) for r in recipients]
        return await asyncio.gather(*tasks)

Shared State Management

class SharedState:
    """
    Distributed state management for agents
    """
    def __init__(self):
        self.store = {}
        self.locks = {}
        self.subscriptions = defaultdict(list)
        
    async def get(self, key):
        """Read from shared state"""
        return self.store.get(key)
        
    async def set(self, key, value, agent_id):
        """Write to shared state with conflict resolution"""
        async with self.lock(key):
            old_value = self.store.get(key)
            
            # Conflict resolution
            if old_value and old_value.version > value.version:
                value = self.merge(old_value, value)
                
            self.store[key] = value
            
            # Notify subscribers
            await self.notify_subscribers(key, value, agent_id)
            
    async def subscribe(self, key, callback):
        """Subscribe to state changes"""
        self.subscriptions[key].append(callback)

Memory Systems

class AgentMemory:
    """
    Hierarchical memory system for agents
    """
    def __init__(self):
        self.working_memory = WorkingMemory(capacity=7)
        self.episodic_memory = EpisodicMemory()
        self.semantic_memory = SemanticMemory()
        self.procedural_memory = ProceduralMemory()
        
    async def store(self, observation, thought, action):
        """Store experience in appropriate memory"""
        episode = Episode(observation, thought, action)
        
        # Working memory (immediate)
        self.working_memory.add(episode)
        
        # Episodic memory (experiences)
        if episode.is_significant():
            await self.episodic_memory.store(episode)
            
        # Semantic memory (facts)
        facts = self.extract_facts(episode)
        await self.semantic_memory.update(facts)
        
        # Procedural memory (skills)
        if episode.demonstrates_skill():
            await self.procedural_memory.learn(episode)
            
    async def recall(self, query=None):
        """Retrieve relevant memories"""
        if query:
            return await self.semantic_memory.search(query)
        else:
            return self.working_memory.get_all()

Tool Integration (MCP)

class MCPToolAdapter:
    """
    Model Context Protocol tool integration
    """
    def __init__(self):
        self.mcp_client = MCPClient()
        self.tool_registry = {}
        
    async def register_tool(self, tool_spec):
        """Register MCP-compatible tool"""
        tool = MCPTool(tool_spec)
        self.tool_registry[tool.name] = tool
        return tool
        
    async def execute(self, tool_name, params):
        """Execute tool via MCP"""
        tool = self.tool_registry[tool_name]
        
        # Validate parameters
        validated = tool.validate_params(params)
        
        # Execute via MCP
        result = await self.mcp_client.execute(
            tool=tool,
            params=validated
        )
        
        return result

Observability

class AgentObserver:
    """
    Monitoring and debugging for agents
    """
    def __init__(self):
        self.traces = []
        self.metrics = MetricsCollector()
        self.logs = LogCollector()
        
    def trace_thought(self, agent_id, thought):
        """Trace agent reasoning"""
        self.traces.append({
            "timestamp": time.time(),
            "agent_id": agent_id,
            "type": "thought",
            "content": thought,
            "tokens_used": thought.token_count
        })
        
    def record_metric(self, name, value, tags=None):
        """Record performance metrics"""
        self.metrics.record(name, value, tags)
        
    async def export_traces(self):
        """Export traces for analysis"""
        return {
            "traces": self.traces,
            "metrics": await self.metrics.aggregate(),
            "logs": self.logs.get_recent()
        }

Deployment Patterns

Standalone Agent

# Single agent for specific task
agent = HanzoAgent(
    config=AgentConfig(
        model="gpt-4",
        tools=["search", "calculator"],
        memory_size="1GB"
    )
)

result = await agent.execute_task(
    "Research quantum computing applications"
)

Agent Swarm

# Multiple collaborative agents
orchestrator = AgentOrchestrator()

# Spawn specialized agents
orchestrator.spawn_agent("researcher", ResearchConfig())
orchestrator.spawn_agent("writer", WriterConfig())
orchestrator.spawn_agent("editor", EditorConfig())

# Execute complex task
result = await orchestrator.delegate_task(
    "Write a comprehensive report on AI safety"
)

Agent Network

# Distributed agent network
network = AgentNetwork(
    discovery_service="hanzo://discovery.hanzo.ai",
    network_id="mainnet"
)

# Join network
agent = HanzoAgent(config)
await network.register(agent)

# Find and collaborate
experts = await network.discover_agents(
    capability="medical_diagnosis"
)

SDK Examples

Python SDK

from hanzoai.agent import Agent, Orchestrator

# Simple agent
agent = Agent(
    name="assistant",
    model="gpt-4",
    instructions="You are a helpful assistant"
)

response = await agent.run("Help me plan a trip to Japan")

# Multi-agent system
orchestrator = Orchestrator()
orchestrator.add_agent("researcher", research_agent)
orchestrator.add_agent("planner", planning_agent)

itinerary = await orchestrator.execute(
    task="Plan a 2-week Japan trip",
    agents=["researcher", "planner"]
)

TypeScript SDK

import { Agent, Orchestrator } from '@hanzoai/agent';

// Create agent
const agent = new Agent({
  name: 'assistant',
  model: 'gpt-4',
  instructions: 'You are a helpful assistant'
});

const response = await agent.run('Help me plan a trip');

// Multi-agent
const orchestrator = new Orchestrator();
orchestrator.addAgent('researcher', researchAgent);
orchestrator.addAgent('planner', plannerAgent);

const result = await orchestrator.execute({
  task: 'Plan a trip',
  agents: ['researcher', 'planner']
});