---
hip: 0009
title: Agent SDK - Multi-Agent Orchestration Framework
author: Hanzo AI Team
type: Standards Track
category: Core
status: Draft
created: 2025-01-09
requires: HIP-1, HIP-2, HIP-4
---


# 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](https://github.com/hanzoai/agent)  
**PyPI**: `hanzoai-agent`  
**NPM**: `@hanzoai/agent`

## Specification

### Agent Architecture

```python
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

```python
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

```yaml
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

```python
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

```python
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

```python
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)

```python
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

```python
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
```python
# 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
```python
# 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
```python
# 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
```python
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
```typescript
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']
});
```

4. [HIP-10: MCP Integration](./hip-10.md)
5. [Agent SDK Repository](https://github.com/hanzoai/agent)

## Copyright

Copyright and related rights waived via [CC0](https://creativecommons.org/publicdomain/zero/1.0/).