Lesson 9: Multi-Agent Systems with OpenShell

Multi-agent systems represent a paradigm shift in AI application development. Instead of building monolithic agents that try to do everything, we decompose complex tasks into specialized agents that collaborate through well-defined interfaces. OpenShell and the NemoClaw blueprint provide the governed runtime environment that makes this architecture production-ready.

1. Why Multi-Agent?

The multi-agent approach addresses fundamental limitations of single-agent systems. According to LangChain's definition, a multi-agent system consists of "multiple independent actors powered by language models connected in a specific way." Each agent maintains its own prompt, LLM, tools, and custom code for collaboration.

Benefits of Multiple Specialized Agents

Division of Responsibilities

Parallel Execution

Multi-agent systems enable concurrent processing of independent subtasks:

2. Isolation Benefits

NVIDIA OpenShell provides the governed runtime environment that makes multi-agent systems secure and manageable. Each agent operates in its own sandbox with defined policies.

Each Agent in Its Own Sandbox

Credential Separation

Blast Radius Containment

If one agent is compromised or malfunctions, the damage is contained to its sandbox:

Resource Isolation

Each sandbox has its own resource limits and quotas:

# OpenShell Resource Configuration
sandbox:
  agent_frontend:
    cpu_limit: "2"
    memory_limit: "4Gi"
    disk_limit: "10Gi"
    network_bandwidth: "100Mbps"

  agent_backend:
    cpu_limit: "4"
    memory_limit: "8Gi"
    disk_limit: "50Gi"
    network_bandwidth: "1Gbps"

  agent_ml:
    cpu_limit: "8"
    memory_limit: "32Gi"
    gpu_access: true
    disk_limit: "100Gi"

3. Architecture Patterns

Multi-agent systems can be organized in several patterns, each suited to different use cases.

Orchestrator Pattern

A central supervisor agent routes tasks to specialized worker agents. The supervisor functions as "an agent whose tools are other agents."

from deepagents import create_deep_agent
from deepagents.middleware import SubAgentMiddleware

# Create specialized agents
research_agent = create_deep_agent(
    model="anthropic:claude-sonnet-4-6",
    tools=[web_search, document_reader],
    system_prompt="You are a research specialist."
)

coding_agent = create_deep_agent(
    model="anthropic:claude-sonnet-4-6",
    tools=[code_executor, file_editor],
    system_prompt="You are a coding specialist."
)

# Orchestrator treats agents as tools
def call_research(query: str) -> str:
    """Delegate research tasks to the research agent."""
    return research_agent.invoke({"messages": [{"role": "user", "content": query}]})

def call_coding(task: str) -> str:
    """Delegate coding tasks to the coding agent."""
    return coding_agent.invoke({"messages": [{"role": "user", "content": task}]})

orchestrator = create_deep_agent(
    model="anthropic:claude-sonnet-4-6",
    tools=[call_research, call_coding],
    system_prompt="""You are an orchestrator.
    Analyze user requests and delegate to the appropriate specialist."""
)

Pipeline Pattern

Agents are arranged in a linear sequence, each processing and passing results to the next:

from typing import TypedDict

class PipelineState(TypedDict):
    raw_input: str
    parsed_data: dict
    analysis: dict
    generated_code: str
    validation_result: dict

async def run_pipeline(user_input: str) -> PipelineState:
    state = {"raw_input": user_input}

    # Stage 1: Parse
    state["parsed_data"] = await parser_agent.invoke(state["raw_input"])

    # Stage 2: Analyze
    state["analysis"] = await analyzer_agent.invoke(state["parsed_data"])

    # Stage 3: Generate
    state["generated_code"] = await generator_agent.invoke(state["analysis"])

    # Stage 4: Validate
    state["validation_result"] = await validator_agent.invoke(state["generated_code"])

    return state

Peer-to-Peer Pattern

Agents communicate directly with each other based on task requirements:

Hierarchical Delegation

Nested teams where managers coordinate sub-teams, which may contain their own sub-agents:

4. Policy Per Agent

OpenShell enables different blueprints and policies for each agent based on their role. This implements the principle of least privilege at the agent level.

Different Blueprints Per Agent

# blueprints/frontend-agent.yaml
name: frontend-agent
version: "1.0"

sandbox:
  type: container
  image: node:20-alpine

filesystem:
  permissions:
    - operations: ["read", "write"]
      paths: ["src/components/**", "src/styles/**"]
      mode: allow
    - operations: ["read"]
      paths: ["src/**"]
      mode: allow
    - operations: ["read", "write"]
      paths: ["**"]
      mode: deny

network:
  allowed_hosts:
    - "cdn.jsdelivr.net"
    - "unpkg.com"
    - "fonts.googleapis.com"
  blocked_hosts:
    - "*.internal.company.com"
    - "database.*"

tools:
  enabled:
    - npm_install
    - npm_build
    - prettier_format
  disabled:
    - execute_sql
    - deploy_production

# blueprints/backend-agent.yaml
name: backend-agent
version: "1.0"

sandbox:
  type: container
  image: python:3.12-slim

filesystem:
  permissions:
    - operations: ["read", "write"]
      paths: ["src/api/**", "src/models/**", "src/services/**"]
      mode: allow
    - operations: ["read"]
      paths: ["src/**", "config/**"]
      mode: allow
    - operations: ["read", "write"]
      paths: ["**"]
      mode: deny

network:
  allowed_hosts:
    - "database.internal.company.com:5432"
    - "redis.internal.company.com:6379"
    - "pypi.org"
  blocked_hosts:
    - "*.external.com"

tools:
  enabled:
    - execute_sql
    - run_migrations
    - pytest
  disabled:
    - npm_install
    - deploy_production

# blueprints/devops-agent.yaml
name: devops-agent
version: "1.0"

sandbox:
  type: container
  image: docker:24-dind
  privileged: true  # Required for Docker-in-Docker

filesystem:
  permissions:
    - operations: ["read", "write"]
      paths: ["infrastructure/**", "k8s/**", ".github/**"]
      mode: allow
    - operations: ["read"]
      paths: ["**"]
      mode: allow

network:
  allowed_hosts:
    - "*.amazonaws.com"
    - "registry.hub.docker.com"
    - "ghcr.io"

secrets:
  accessible:
    - AWS_ACCESS_KEY_ID
    - AWS_SECRET_ACCESS_KEY
    - DOCKER_REGISTRY_TOKEN

tools:
  enabled:
    - deploy_staging
    - deploy_production
    - kubectl
    - terraform
  requires_approval:
    - deploy_production
    - terraform_apply

Role-Based Network Access Diagram

5. Inter-Agent Communication

Agents in a multi-agent system need to share results while maintaining isolation. Here are the primary communication patterns.

How Agents Share Results

All patterns use graph state for communication:

• Agents add to the graph's state
• Control flow managed by edges
• State machines perspective: agents as states, connections as transition matrices

Communication Pattern Comparison

Shared Volumes

Agents can share files through mounted volumes with controlled permissions:

from deepagents import create_deep_agent
from deepagents.filesystem import CompositeFilesystem, LocalBackend

# Create shared workspace
shared_workspace = LocalBackend(
    root="/workspace/shared",
    permissions=[
        {"operations": ["read", "write"], "paths": ["output/**"], "mode": "allow"},
        {"operations": ["read"], "paths": ["input/**"], "mode": "allow"},
    ]
)

# Agent 1: Writes to shared output
agent1 = create_deep_agent(
    model="anthropic:claude-sonnet-4-6",
    filesystem=CompositeFilesystem(
        mounts={
            "/workspace": shared_workspace,
            "/agent1": LocalBackend(root="/sandbox/agent1"),
        }
    ),
    system_prompt="Write analysis results to /workspace/output/"
)

# Agent 2: Reads from shared output
agent2 = create_deep_agent(
    model="anthropic:claude-sonnet-4-6",
    filesystem=CompositeFilesystem(
        mounts={
            "/workspace": shared_workspace,
            "/agent2": LocalBackend(root="/sandbox/agent2"),
        }
    ),
    system_prompt="Read analysis from /workspace/output/ and generate report"
)

Message Passing

from dataclasses import dataclass
from typing import Any
from collections import deque

@dataclass
class AgentMessage:
    sender: str
    recipient: str
    content: Any
    message_type: str  # "request", "response", "notification"
    correlation_id: str

class MessageBus:
    def __init__(self):
        self._queues: dict[str, deque] = {}

    def register_agent(self, agent_id: str):
        self._queues[agent_id] = deque()

    def send(self, message: AgentMessage):
        if message.recipient in self._queues:
            self._queues[message.recipient].append(message)

    def receive(self, agent_id: str) -> AgentMessage | None:
        if agent_id in self._queues and self._queues[agent_id]:
            return self._queues[agent_id].popleft()
        return None

# Usage in orchestrator
bus = MessageBus()
bus.register_agent("research_agent")
bus.register_agent("coding_agent")
bus.register_agent("review_agent")

# Research agent sends findings
bus.send(AgentMessage(
    sender="research_agent",
    recipient="coding_agent",
    content={"api_spec": "...", "examples": [...]},
    message_type="response",
    correlation_id="task-123"
))

Avoiding Direct Access

Agents should never access each other's internal state directly. All communication flows through controlled channels:

6. Scaling Considerations

Resource Management

Multi-agent systems require careful resource planning:

# Resource allocation strategy
scaling:
  strategy: dynamic

  agent_pools:
    research:
      min_instances: 1
      max_instances: 5
      scale_on:
        - metric: queue_depth
          threshold: 10
          action: scale_up
        - metric: idle_time
          threshold: 300s
          action: scale_down

    coding:
      min_instances: 2
      max_instances: 10
      scale_on:
        - metric: cpu_utilization
          threshold: 80%
          action: scale_up

    review:
      min_instances: 1
      max_instances: 3
      # Reviews are less frequent but critical
      priority: high

  resource_quotas:
    total_cpu: "32"
    total_memory: "128Gi"
    total_gpu: "4"

  cost_limits:
    daily_llm_budget: 100.00  # USD
    alert_threshold: 0.8  # Alert at 80%

Concurrent Sandbox Limits

Cost Optimization Strategies

from dataclasses import dataclass
from enum import Enum

class ModelTier(Enum):
    FAST = "claude-3-haiku"      # $0.25/M input, $1.25/M output
    BALANCED = "claude-sonnet-4-6"   # $3/M input, $15/M output
    POWERFUL = "claude-opus-4"     # $15/M input, $75/M output

@dataclass
class AgentCostProfile:
    agent_type: str
    model_tier: ModelTier
    estimated_tokens_per_task: int

cost_profiles = {
    "triage": AgentCostProfile("triage", ModelTier.FAST, 1000),
    "research": AgentCostProfile("research", ModelTier.BALANCED, 5000),
    "coding": AgentCostProfile("coding", ModelTier.BALANCED, 10000),
    "review": AgentCostProfile("review", ModelTier.POWERFUL, 3000),
    "orchestrator": AgentCostProfile("orchestrator", ModelTier.FAST, 500),
}

def select_model_for_task(task_complexity: str, budget_remaining: float) -> ModelTier:
    """Dynamic model selection based on task and budget."""
    if budget_remaining < 10.0:
        return ModelTier.FAST

    complexity_map = {
        "simple": ModelTier.FAST,
        "medium": ModelTier.BALANCED,
        "complex": ModelTier.POWERFUL,
    }
    return complexity_map.get(task_complexity, ModelTier.BALANCED)

7. Complete Example: Multi-Agent Coding System

Let's build a complete multi-agent system for software development tasks.

System Architecture

Blueprint Configurations

# blueprints/orchestrator.yaml
name: orchestrator-agent
version: "1.0"

model:
  provider: anthropic
  name: claude-3-haiku  # Fast model for routing decisions
  max_tokens: 1000

sandbox:
  type: minimal

filesystem:
  permissions:
    - operations: ["read"]
      paths: ["**"]
      mode: allow
    - operations: ["write"]
      paths: ["orchestration/**"]
      mode: allow

network:
  allowed_hosts: []  # No network access needed

tools:
  enabled:
    - delegate_to_architect
    - delegate_to_coder
    - delegate_to_reviewer
    - delegate_to_tester
    - read_shared_files
    - write_final_result

# blueprints/architect.yaml
name: architect-agent
version: "1.0"

model:
  provider: anthropic
  name: claude-sonnet-4-6
  max_tokens: 4000

sandbox:
  type: container
  image: python:3.12-slim

filesystem:
  permissions:
    - operations: ["read"]
      paths: ["src/**", "docs/**", "requirements.txt"]
      mode: allow
    - operations: ["write"]
      paths: ["shared/designs/**"]
      mode: allow

network:
  allowed_hosts:
    - "api.github.com"  # For reading existing patterns

tools:
  enabled:
    - analyze_codebase
    - create_design_doc
    - diagram_generator

# blueprints/coder.yaml
name: coding-agent
version: "1.0"

model:
  provider: anthropic
  name: claude-sonnet-4-6
  max_tokens: 8000

sandbox:
  type: container
  image: python:3.12

filesystem:
  permissions:
    - operations: ["read"]
      paths: ["**"]
      mode: allow
    - operations: ["write"]
      paths: ["src/**", "tests/**", "shared/code/**"]
      mode: allow
    - operations: ["write"]
      paths: ["config/**", ".env*"]
      mode: deny

network:
  allowed_hosts:
    - "pypi.org"
    - "files.pythonhosted.org"

tools:
  enabled:
    - read_file
    - write_file
    - edit_file
    - execute
    - pip_install
    - run_linter

# blueprints/reviewer.yaml
name: review-agent
version: "1.0"

model:
  provider: anthropic
  name: claude-opus-4  # Most capable for thorough review
  max_tokens: 4000

sandbox:
  type: container
  image: python:3.12-slim

filesystem:
  permissions:
    - operations: ["read"]
      paths: ["**"]
      mode: allow
    - operations: ["write"]
      paths: ["shared/reviews/**"]
      mode: allow

network:
  allowed_hosts: []

tools:
  enabled:
    - read_file
    - analyze_diff
    - security_scan
    - write_review

# blueprints/tester.yaml
name: test-agent
version: "1.0"

model:
  provider: anthropic
  name: claude-sonnet-4-6
  max_tokens: 4000

sandbox:
  type: container
  image: python:3.12

filesystem:
  permissions:
    - operations: ["read"]
      paths: ["**"]
      mode: allow
    - operations: ["write"]
      paths: ["tests/**", "shared/test_results/**"]
      mode: allow

network:
  allowed_hosts:
    - "pypi.org"

tools:
  enabled:
    - read_file
    - write_file
    - execute
    - pytest
    - coverage_report

Python Orchestration Code

"""
Multi-Agent Coding System with OpenShell
"""
from dataclasses import dataclass, field
from enum import Enum
from typing import Any
import asyncio
import uuid

from deepagents import create_deep_agent
from deepagents.middleware import FilesystemMiddleware, SandboxMiddleware
from deepagents.filesystem import LocalBackend, CompositeFilesystem


class TaskStatus(Enum):
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"


@dataclass
class Task:
    id: str
    type: str
    description: str
    status: TaskStatus = TaskStatus.PENDING
    result: Any = None
    dependencies: list[str] = field(default_factory=list)


@dataclass
class AgentResult:
    agent_name: str
    task_id: str
    success: bool
    output: Any
    artifacts: list[str] = field(default_factory=list)


class MultiAgentCodingSystem:
    """Orchestrates multiple specialized agents for software development."""

    def __init__(self, workspace_path: str):
        self.workspace = workspace_path
        self.shared_fs = self._create_shared_filesystem()
        self.agents = self._initialize_agents()
        self.tasks: dict[str, Task] = {}
        self.results: list[AgentResult] = []

    def _create_shared_filesystem(self) -> CompositeFilesystem:
        """Create filesystem with proper isolation."""
        return CompositeFilesystem(
            mounts={
                "/workspace": LocalBackend(root=self.workspace),
                "/shared": LocalBackend(
                    root=f"{self.workspace}/shared",
                    permissions=[
                        {"operations": ["read", "write"], "paths": ["**"], "mode": "allow"}
                    ]
                ),
            }
        )

    def _initialize_agents(self) -> dict:
        """Initialize all specialized agents."""

        # Architect Agent - designs solutions
        architect = create_deep_agent(
            model="anthropic:claude-sonnet-4-6",
            system_prompt="""You are a software architect. Your responsibilities:
            1. Analyze requirements and existing codebase
            2. Create design documents with clear specifications
            3. Define interfaces and data structures
            4. Write designs to /shared/designs/

            Focus on clarity, maintainability, and best practices.""",
            middleware=[
                FilesystemMiddleware(
                    backend=self.shared_fs,
                    tools=["read_file", "write_file", "ls", "glob"]
                ),
            ],
        )

        # Coding Agent - implements solutions
        coder = create_deep_agent(
            model="anthropic:claude-sonnet-4-6",
            system_prompt="""You are a senior software developer. Your responsibilities:
            1. Read design documents from /shared/designs/
            2. Implement code following the specifications
            3. Write clean, well-documented code
            4. Save implementations to /workspace/src/

            Follow existing code patterns and conventions.""",
            middleware=[
                FilesystemMiddleware(
                    backend=self.shared_fs,
                    tools=["read_file", "write_file", "edit_file", "ls", "glob", "grep"]
                ),
                SandboxMiddleware(
                    image="python:3.12",
                    tools=["execute"],
                    network_allowed=["pypi.org"]
                ),
            ],
        )

        # Review Agent - reviews code quality
        reviewer = create_deep_agent(
            model="anthropic:claude-opus-4",
            system_prompt="""You are a code reviewer. Your responsibilities:
            1. Review code for correctness, security, and best practices
            2. Check alignment with design documents
            3. Identify potential bugs and improvements
            4. Write review comments to /shared/reviews/

            Be thorough but constructive in feedback.""",
            middleware=[
                FilesystemMiddleware(
                    backend=self.shared_fs,
                    tools=["read_file", "write_file", "ls", "glob", "grep"]
                ),
            ],
        )

        # Test Agent - writes and runs tests
        tester = create_deep_agent(
            model="anthropic:claude-sonnet-4-6",
            system_prompt="""You are a test engineer. Your responsibilities:
            1. Write comprehensive unit and integration tests
            2. Execute tests and report results
            3. Ensure adequate code coverage
            4. Save tests to /workspace/tests/
            5. Write results to /shared/test_results/

            Aim for edge cases and failure scenarios.""",
            middleware=[
                FilesystemMiddleware(
                    backend=self.shared_fs,
                    tools=["read_file", "write_file", "edit_file", "ls", "glob"]
                ),
                SandboxMiddleware(
                    image="python:3.12",
                    tools=["execute"],
                    network_allowed=["pypi.org"]
                ),
            ],
        )

        return {
            "architect": architect,
            "coder": coder,
            "reviewer": reviewer,
            "tester": tester,
        }

    async def execute_task(self, agent_name: str, task: Task) -> AgentResult:
        """Execute a task with the specified agent."""
        agent = self.agents[agent_name]
        task.status = TaskStatus.IN_PROGRESS

        try:
            response = await agent.ainvoke({
                "messages": [{"role": "user", "content": task.description}]
            })

            task.status = TaskStatus.COMPLETED
            task.result = response

            result = AgentResult(
                agent_name=agent_name,
                task_id=task.id,
                success=True,
                output=response,
            )

        except Exception as e:
            task.status = TaskStatus.FAILED
            task.result = str(e)

            result = AgentResult(
                agent_name=agent_name,
                task_id=task.id,
                success=False,
                output=str(e),
            )

        self.results.append(result)
        return result

    async def run_development_workflow(self, requirement: str) -> dict:
        """Run the complete development workflow."""

        # Phase 1: Architecture
        design_task = Task(
            id=str(uuid.uuid4()),
            type="design",
            description=f"""Create a design document for the following requirement:

{requirement}

Include:
1. High-level architecture
2. Component breakdown
3. Interface definitions
4. Data structures
5. Implementation plan

Save the design to /shared/designs/design.md"""
        )
        self.tasks[design_task.id] = design_task
        design_result = await self.execute_task("architect", design_task)

        if not design_result.success:
            return {"status": "failed", "phase": "design", "error": design_result.output}

        # Phase 2: Implementation
        code_task = Task(
            id=str(uuid.uuid4()),
            type="implementation",
            description="""Read the design document from /shared/designs/design.md

Implement the solution according to the specifications.
- Create all necessary files in /workspace/src/
- Follow the interfaces defined in the design
- Add appropriate docstrings and comments
- Handle errors gracefully""",
            dependencies=[design_task.id]
        )
        self.tasks[code_task.id] = code_task
        code_result = await self.execute_task("coder", code_task)

        if not code_result.success:
            return {"status": "failed", "phase": "implementation", "error": code_result.output}

        # Phase 3: Testing (can run in parallel with review)
        test_task = Task(
            id=str(uuid.uuid4()),
            type="testing",
            description="""Read the implementation from /workspace/src/

Write comprehensive tests:
1. Unit tests for all functions
2. Edge case testing
3. Error handling tests

Save tests to /workspace/tests/
Run the tests and save results to /shared/test_results/results.md""",
            dependencies=[code_task.id]
        )

        # Phase 3b: Code Review (parallel with testing)
        review_task = Task(
            id=str(uuid.uuid4()),
            type="review",
            description="""Review the implementation:

1. Read design from /shared/designs/design.md
2. Read implementation from /workspace/src/
3. Check for:
   - Correctness against design
   - Security issues
   - Performance concerns
   - Code quality and style
   - Error handling

Save review to /shared/reviews/review.md""",
            dependencies=[code_task.id]
        )

        self.tasks[test_task.id] = test_task
        self.tasks[review_task.id] = review_task

        # Run testing and review in parallel
        test_result, review_result = await asyncio.gather(
            self.execute_task("tester", test_task),
            self.execute_task("reviewer", review_task)
        )

        return {
            "status": "completed",
            "results": {
                "design": design_result.output,
                "implementation": code_result.output,
                "tests": test_result.output,
                "review": review_result.output,
            },
            "artifacts": [
                "/shared/designs/design.md",
                "/workspace/src/",
                "/workspace/tests/",
                "/shared/test_results/results.md",
                "/shared/reviews/review.md",
            ]
        }


# Usage example
async def main():
    system = MultiAgentCodingSystem(workspace_path="/projects/my-app")

    result = await system.run_development_workflow(
        requirement="""
        Build a REST API endpoint for user authentication that:
        1. Accepts email and password
        2. Validates credentials against a database
        3. Returns a JWT token on success
        4. Implements rate limiting
        5. Logs authentication attempts
        """
    )

    print(f"Workflow completed with status: {result['status']}")
    if result['status'] == 'completed':
        print(f"Artifacts created: {result['artifacts']}")


if __name__ == "__main__":
    asyncio.run(main())

Execution Flow Diagram

Knowledge Check

Summary