add multiple agents

Signed-off-by: Jintao Zhang <zhangjintao9020@gmail.com>

src/amcp/__init__.py

@@ -1,2 +1,56 @@
-__all__ = ["__version__"]
-__version__ = "0.1.0"
+"""AMCP - Lego-style Coding Agent CLI with Multi-Agent Support."""
+
+__all__ = [
+    "__version__",
+    # Agent classes and functions
+    "Agent",
+    "AgentExecutionError",
+    "MaxStepsReached",
+    "BusyError",
+    "create_agent_by_name",
+    "create_agent_from_config",
+    "create_subagent",
+    "list_available_agents",
+    "list_primary_agents",
+    "list_subagent_types",
+    # Multi-agent system
+    "AgentMode",
+    "AgentConfig",
+    "AgentRegistry",
+    "get_agent_registry",
+    "get_agent_config",
+    # Message queue
+    "MessagePriority",
+    "QueuedMessage",
+    "MessageQueueManager",
+    "get_message_queue_manager",
+]
+
+__version__ = "0.2.0"
+
+# Lazy imports for cleaner namespace
+from .agent import (
+    Agent,
+    AgentExecutionError,
+    BusyError,
+    MaxStepsReached,
+    create_agent_by_name,
+    create_agent_from_config,
+    create_subagent,
+    list_available_agents,
+    list_primary_agents,
+    list_subagent_types,
+)
+from .message_queue import (
+    MessagePriority,
+    MessageQueueManager,
+    QueuedMessage,
+    get_message_queue_manager,
+)
+from .multi_agent import (
+    AgentConfig,
+    AgentMode,
+    AgentRegistry,
+    get_agent_config,
+    get_agent_registry,
+)

src/amcp/agent.py

@@ -14,6 +14,8 @@ from .chat import _make_client, _resolve_api_key, _resolve_base_url
 from .compaction import Compactor
 from .config import load_config
 from .mcp_client import call_mcp_tool, list_mcp_tools
+from .message_queue import MessagePriority, get_message_queue_manager
+from .multi_agent import AgentMode
 from .tools import ToolRegistry
 from .ui import LiveUI
 
@@ -30,6 +32,12 @@ class MaxStepsReached(Exception):
     pass
 
 
+class BusyError(Exception):
+    """Raised when agent session is busy processing another request."""
+
+    pass
+
+
 class Agent:
     """
     Enhanced agent execution engine with tool calling and conversation management.
@@ -222,7 +230,13 @@ class Agent:
         }
 
     async def run(
-        self, user_input: str, work_dir: Path | None = None, stream: bool = True, show_progress: bool = True
+        self,
+        user_input: str,
+        work_dir: Path | None = None,
+        stream: bool = True,
+        show_progress: bool = True,
+        priority: MessagePriority = MessagePriority.NORMAL,
+        queue_if_busy: bool = True,
     ) -> str:
         """
         Run the agent with the given user input.
@@ -232,6 +246,8 @@ class Agent:
             work_dir: Working directory for context
             stream: Whether to stream responses
             show_progress: Whether to show progress indicators
+            priority: Message priority (for queuing)
+            queue_if_busy: Whether to queue the message if session is busy
 
         Returns:
             Agent's response
@@ -239,6 +255,77 @@ class Agent:
         Raises:
             AgentExecutionError: If execution fails
             MaxStepsReached: If max steps exceeded
+            BusyError: If session is busy and queue_if_busy is False
+        """
+        queue_manager = get_message_queue_manager()
+
+        # Check if session is busy
+        if queue_manager.is_busy(self.session_id):
+            if queue_if_busy:
+                # Queue the message for later processing
+                await queue_manager.enqueue(
+                    session_id=self.session_id,
+                    prompt=user_input,
+                    priority=priority,
+                    work_dir=str(work_dir) if work_dir else None,
+                    stream=stream,
+                    show_progress=show_progress,
+                )
+                self.console.print(
+                    f"[dim]Message queued ({queue_manager.queued_count(self.session_id)} in queue)[/dim]"
+                )
+                return "[Message queued for later processing]"
+            else:
+                raise BusyError(f"Session {self.session_id} is busy processing another request")
+
+        # Acquire session lock
+        acquired = await queue_manager.acquire(self.session_id)
+        if not acquired:
+            # Race condition - queue it
+            if queue_if_busy:
+                await queue_manager.enqueue(
+                    session_id=self.session_id,
+                    prompt=user_input,
+                    priority=priority,
+                )
+                return "[Message queued for later processing]"
+            else:
+                raise BusyError(f"Session {self.session_id} is busy processing another request")
+
+        try:
+            # Process the current message
+            result = await self._process_message(user_input, work_dir, stream, show_progress)
+
+            # Process any queued messages
+            while True:
+                next_msg = await queue_manager.dequeue(self.session_id)
+                if not next_msg:
+                    break
+
+                # Process queued message
+                self.console.print(f"[dim]Processing queued message...[/dim]")
+                queued_work_dir = Path(next_msg.metadata.get("work_dir")) if next_msg.metadata.get("work_dir") else work_dir
+                await self._process_message(
+                    next_msg.prompt,
+                    queued_work_dir,
+                    next_msg.metadata.get("stream", stream),
+                    next_msg.metadata.get("show_progress", show_progress),
+                )
+
+            return result
+
+        finally:
+            # Always release the session lock
+            queue_manager.release(self.session_id)
+
+    async def _process_message(
+        self, user_input: str, work_dir: Path | None, stream: bool, show_progress: bool
+    ) -> str:
+        """
+        Process a single message (internal implementation).
+
+        This is the core message processing logic, extracted from run()
+        to support queue-based processing.
         """
         # Save user input to conversation history immediately to preserve context
         self.conversation_history.append({"role": "user", "content": user_input})
@@ -247,10 +334,6 @@ class Agent:
             with self._create_progress_context(show_progress) as status:
                 status.update(f"[bold]Agent {self.name}[/bold] thinking...")
 
-                # DO NOT reset current conversation tool calls counter here
-                # It should persist across multiple calls within the same session
-                # Reset only when explicitly requested or when starting a completely new session
-
                 # Prepare messages with conversation history
                 system_prompt = self._get_system_prompt(work_dir)
                 messages = [{"role": "system", "content": system_prompt}]
@@ -301,6 +384,27 @@ class Agent:
             self.console.print(f"[red]Agent execution failed:[/red] {e}")
             raise AgentExecutionError(f"Agent execution failed: {e}") from e
 
+    def is_busy(self) -> bool:
+        """Check if this agent's session is currently busy."""
+        return get_message_queue_manager().is_busy(self.session_id)
+
+    def queued_count(self) -> int:
+        """Get the number of queued messages for this session."""
+        return get_message_queue_manager().queued_count(self.session_id)
+
+    def queued_prompts(self) -> list[str]:
+        """Get list of queued prompts for this session."""
+        return get_message_queue_manager().queued_prompts(self.session_id)
+
+    async def clear_queue(self) -> int:
+        """Clear all queued messages for this session."""
+        return await get_message_queue_manager().clear_queue(self.session_id)
+
+    def get_queue_status(self) -> dict[str, Any]:
+        """Get queue status for this session."""
+        return get_message_queue_manager().get_queue_status(self.session_id)
+
+
     async def _build_tools(self) -> list[dict[str, Any]]:
         """Build list of available tools."""
         tools = []
@@ -626,9 +730,151 @@ class Agent:
         """Get summary of agent execution."""
         return {
             "agent_name": self.name,
+            "agent_mode": self.agent_spec.mode.value,
             "steps_taken": self.step_count,
             "max_steps": self.max_steps,
             "tools_called": len(self.tool_calls_history),
             "tool_calls": self.tool_calls_history,
             "context_vars": self._get_context_vars(),
+            "can_delegate": self.agent_spec.can_delegate,
+            "is_busy": self.is_busy(),
+            "queued_count": self.queued_count(),
         }
+
+
+# Factory functions for creating agents from multi-agent configurations
+
+
+def create_agent_from_config(
+    config: "AgentConfig",
+    session_id: str | None = None,
+) -> Agent:
+    """
+    Create an Agent from an AgentConfig.
+
+    This is the primary way to instantiate agents using the multi-agent system.
+
+    Args:
+        config: AgentConfig from the multi_agent module
+        session_id: Optional session ID for conversation persistence
+
+    Returns:
+        Configured Agent instance
+    """
+    from .multi_agent import AgentConfig
+
+    # Convert AgentConfig to ResolvedAgentSpec
+    from .agent_spec import ResolvedAgentSpec
+
+    spec = ResolvedAgentSpec(
+        name=config.name,
+        description=config.description,
+        mode=config.mode,
+        system_prompt=config.system_prompt,
+        tools=config.tools,
+        exclude_tools=config.excluded_tools,
+        max_steps=config.max_steps,
+        model="",  # Use default from config
+        base_url="",  # Use default from config
+        can_delegate=config.can_delegate,
+    )
+
+    return Agent(agent_spec=spec, session_id=session_id)
+
+
+def create_agent_by_name(
+    name: str,
+    session_id: str | None = None,
+) -> Agent:
+    """
+    Create an Agent by looking up its name in the registry.
+
+    Args:
+        name: Name of the agent in the registry (e.g., "coder", "explorer", "planner")
+        session_id: Optional session ID for conversation persistence
+
+    Returns:
+        Configured Agent instance
+
+    Raises:
+        ValueError: If agent name is not found in registry
+    """
+    from .multi_agent import get_agent_config
+
+    config = get_agent_config(name)
+    if config is None:
+        from .multi_agent import get_agent_registry
+        available = get_agent_registry().list_agents()
+        raise ValueError(f"Unknown agent: {name}. Available agents: {', '.join(available)}")
+
+    return create_agent_from_config(config, session_id)
+
+
+def create_subagent(
+    parent_agent: Agent,
+    task_description: str,
+    tools: list[str] | None = None,
+) -> Agent:
+    """
+    Create a subagent for a specific task.
+
+    This creates a new agent that inherits the session from the parent
+    but has a focused task and possibly restricted tools.
+
+    Args:
+        parent_agent: The parent agent creating this subagent
+        task_description: Description of the task for the subagent
+        tools: Optional list of tools for the subagent
+
+    Returns:
+        New Agent configured as a subagent
+
+    Raises:
+        ValueError: If parent agent cannot delegate
+    """
+    if not parent_agent.agent_spec.can_delegate:
+        raise ValueError(f"Agent '{parent_agent.name}' cannot delegate to subagents")
+
+    from .multi_agent import create_subagent_config
+
+    config = create_subagent_config(
+        parent_name=parent_agent.name,
+        task_description=task_description,
+        tools=tools,
+    )
+
+    # Create subagent with a new session (isolated from parent)
+    return create_agent_from_config(config)
+
+
+def list_available_agents() -> list[str]:
+    """
+    List all available agent names.
+
+    Returns:
+        List of agent names from the registry
+    """
+    from .multi_agent import get_agent_registry
+    return get_agent_registry().list_agents()
+
+
+def list_primary_agents() -> list[str]:
+    """
+    List all primary (main) agent names.
+
+    Returns:
+        List of primary agent names
+    """
+    from .multi_agent import get_agent_registry
+    return get_agent_registry().list_primary_agents()
+
+
+def list_subagent_types() -> list[str]:
+    """
+    List all available subagent types.
+
+    Returns:
+        List of subagent names
+    """
+    from .multi_agent import get_agent_registry
+    return get_agent_registry().list_subagents()

src/amcp/agent_spec.py

@@ -7,6 +7,7 @@ import yaml
 from pydantic import BaseModel, Field
 
 from .config import load_config as load_app_config
+from .multi_agent import AgentMode
 
 
 class AgentSpecError(Exception):
@@ -20,6 +21,7 @@ class AgentSpec(BaseModel):
 
     name: str = Field(description="Agent name")
     description: str = Field(default="", description="Agent description")
+    mode: str = Field(default="primary", description="Agent mode: 'primary' or 'subagent'")
     system_prompt: str = Field(description="System prompt for the agent")
     system_prompt_template: str = Field(default="", description="System prompt template with variables")
     system_prompt_vars: dict[str, str] = Field(default_factory=dict, description="Variables for system prompt template")
@@ -28,6 +30,7 @@ class AgentSpec(BaseModel):
     max_steps: int = Field(default=5, description="Maximum tool execution steps")
     model: str = Field(default="", description="Preferred model name")
     base_url: str = Field(default="", description="Preferred base URL")
+    can_delegate: bool = Field(default=True, description="Whether agent can spawn subagents")
 
     class Config:
         extra = "allow"
@@ -39,12 +42,14 @@ class ResolvedAgentSpec:
 
     name: str
     description: str
+    mode: AgentMode
     system_prompt: str
     tools: list[str]
     exclude_tools: list[str]
     max_steps: int
     model: str
     base_url: str
+    can_delegate: bool = True
 
 
 def load_agent_spec(agent_file: Path) -> ResolvedAgentSpec:
@@ -89,15 +94,25 @@ def load_agent_spec(agent_file: Path) -> ResolvedAgentSpec:
     elif spec.system_prompt_template:
         system_prompt = spec.system_prompt_template
 
+    # Parse mode
+    mode = AgentMode.PRIMARY if spec.mode == "primary" else AgentMode.SUBAGENT
+
+    # Subagents cannot delegate by default
+    can_delegate = spec.can_delegate
+    if mode == AgentMode.SUBAGENT:
+        can_delegate = False
+
     return ResolvedAgentSpec(
         name=spec.name,
         description=spec.description,
+        mode=mode,
         system_prompt=system_prompt,
         tools=spec.tools or [],
         exclude_tools=spec.exclude_tools or [],
         max_steps=spec.max_steps,
         model=spec.model or default_model,
         base_url=spec.base_url or default_base_url,
+        can_delegate=can_delegate,
     )
 
 
@@ -106,6 +121,7 @@ def get_default_agent_spec() -> ResolvedAgentSpec:
     return ResolvedAgentSpec(
         name="default",
         description="Default AMCP agent",
+        mode=AgentMode.PRIMARY,
         system_prompt="""You are AMCP, a Lego-style coding agent CLI. You help users with software engineering tasks using the available tools.
 
 Available tools:
@@ -129,6 +145,7 @@ Current time: {current_time}""",
         max_steps=300,
         model="",
         base_url="",
+        can_delegate=True,
     )
 
 

src/amcp/cli.py

@@ -13,11 +13,13 @@ from rich.console import Console
 from rich.json import JSON
 from rich.markdown import Markdown
 from rich.panel import Panel
+from rich.table import Table
 
-from .agent import Agent
+from .agent import Agent, create_agent_by_name
 from .agent_spec import get_default_agent_spec, list_available_agents, load_agent_spec
 from .config import AMCPConfig, load_config, save_default_config
 from .mcp_client import call_mcp_tool, list_mcp_tools
+from .multi_agent import get_agent_registry
 
 app = typer.Typer(add_completion=False, context_settings={"help_option_names": ["-h", "--help"]})
 console = Console()
@@ -101,6 +103,10 @@ def main(
     ctx: typer.Context,
     message: Annotated[str | None, typer.Option("--once", help="Send one message and exit")] = None,
     agent_file: Annotated[str | None, typer.Option("--agent", help="Path to agent specification file")] = None,
+    agent_type: Annotated[
+        str | None,
+        typer.Option("--agent-type", "-t", help="Built-in agent type: coder, explorer, planner, focused_coder"),
+    ] = None,
     work_dir: Annotated[
         Path | None,
         typer.Option(
@@ -109,6 +115,9 @@ def main(
     ] = None,
     no_progress: Annotated[bool, typer.Option("--no-progress", help="Disable progress indicators")] = False,
     list_agents: Annotated[bool, typer.Option("--list", help="List available agent specifications")] = False,
+    list_agent_types: Annotated[
+        bool, typer.Option("--list-types", help="List available built-in agent types")
+    ] = False,
     session_id: Annotated[
         str | None, typer.Option("--session", help="Use specific session ID for conversation continuity")
     ] = None,
@@ -123,6 +132,33 @@ def main(
     if ctx.invoked_subcommand is not None:
         return
 
+    # Handle list agent types
+    if list_agent_types:
+        registry = get_agent_registry()
+        table = Table(title="Available Built-in Agent Types")
+        table.add_column("Name", style="cyan")
+        table.add_column("Mode", style="magenta")
+        table.add_column("Description", style="green")
+        table.add_column("Can Delegate", style="yellow")
+        table.add_column("Max Steps", style="blue")
+
+        for name in registry.list_agents():
+            config = registry.get(name)
+            if config:
+                table.add_row(
+                    config.name,
+                    config.mode.value,
+                    config.description[:50] + "..." if len(config.description) > 50 else config.description,
+                    "✅" if config.can_delegate else "❌",
+                    str(config.max_steps),
+                )
+
+        console.print(table)
+        console.print()
+        console.print("[dim]Use --agent-type <name> or -t <name> to select an agent type[/dim]")
+        console.print("[dim]Example: amcp -t explorer --once 'Find all TODO comments'[/dim]")
+        return
+
     # Handle session listing
     if list_sessions:
         sessions_dir = Path.home() / ".config" / "amcp" / "sessions"
@@ -155,11 +191,11 @@ def main(
         agents_dir = Path(os.path.expanduser("~/.config/amcp/agents"))
         local_agents_dir = Path("agents")
 
-        agent_files = list_available_agents(agents_dir)
+        agent_files_list = list_available_agents(agents_dir)
         local_agent_files = list_available_agents(local_agents_dir)
 
         # Combine both lists
-        all_agent_files = agent_files + local_agent_files
+        all_agent_files = agent_files_list + local_agent_files
 
         if not all_agent_files:
             console.print(
@@ -168,41 +204,71 @@ def main(
             return
 
         console.print("[bold]Available Agent Specifications:[/bold]")
-        for agent_file in all_agent_files:
+        for agent_file_path in all_agent_files:
             try:
-                spec = load_agent_spec(agent_file)
-                console.print(f"📄 {agent_file.name}")
+                spec = load_agent_spec(agent_file_path)
+                console.print(f"📄 {agent_file_path.name}")
                 console.print(f"   Name: {spec.name}")
+                console.print(f"   Mode: {spec.mode.value}")
                 console.print(f"   Description: {spec.description}")
                 console.print(f"   Tools: {len(spec.tools)}")
                 console.print()
             except Exception as e:
-                console.print(f"❌ {agent_file.name}: {e}")
+                console.print(f"❌ {agent_file_path.name}: {e}")
 
         # Also show default agent
         default_spec = get_default_agent_spec()
         console.print("[bold]Default Agent:[/bold]")
         console.print(f"   Name: {default_spec.name}")
+        console.print(f"   Mode: {default_spec.mode.value}")
         console.print(f"   Description: {default_spec.description}")
         console.print(f"   Tools: {len(default_spec.tools)}")
         return
 
+    # Load configuration
+    cfg = load_config()
+
     try:
-        # Load agent specification
+        # Determine agent to use (priority: --agent > --agent-type > config.default_agent > default)
+        agent = None
+
         if agent_file:
+            # Load from YAML file
             agent_path = Path(agent_file).expanduser()
             if not agent_path.exists():
                 console.print(f"[red]Agent file not found: {agent_path}[/red]")
                 raise typer.Exit(1)
             agent_spec = load_agent_spec(agent_path)
-            console.print(f"[green]Loaded agent: {agent_spec.name}[/green]")
+            agent = Agent(agent_spec, session_id=session_id)
+            console.print(f"[green]Loaded agent from file: {agent_spec.name} ({agent_spec.mode.value})[/green]")
+
+        elif agent_type:
+            # Use built-in agent type
+            registry = get_agent_registry()
+            if agent_type not in registry.list_agents():
+                console.print(f"[red]Unknown agent type: {agent_type}[/red]")
+                console.print(f"[dim]Available types: {', '.join(registry.list_agents())}[/dim]")
+                raise typer.Exit(1)
+            agent = create_agent_by_name(agent_type, session_id=session_id)
+            console.print(f"[green]Using agent: {agent.name} ({agent.agent_spec.mode.value})[/green]")
+
+        elif cfg.chat and cfg.chat.default_agent:
+            # Use agent from config
+            registry = get_agent_registry()
+            if cfg.chat.default_agent in registry.list_agents():
+                agent = create_agent_by_name(cfg.chat.default_agent, session_id=session_id)
+                console.print(f"[green]Using configured agent: {agent.name} ({agent.agent_spec.mode.value})[/green]")
+            else:
+                console.print(f"[yellow]Warning: Configured agent '{cfg.chat.default_agent}' not found, using default[/yellow]")
+                agent_spec = get_default_agent_spec()
+                agent = Agent(agent_spec, session_id=session_id)
+
         else:
+            # Use default agent
             agent_spec = get_default_agent_spec()
+            agent = Agent(agent_spec, session_id=session_id)
             console.print(f"[green]Using default agent: {agent_spec.name}[/green]")
 
-        # Create agent with session management
-        agent = Agent(agent_spec, session_id=session_id)
-
         # Handle session clearing
         if clear_session:
             agent.clear_conversation_history()
@@ -233,8 +299,8 @@ def main(
         else:
             # Interactive mode
             console.print(f"[bold]🤖 Agent {agent.name} - Interactive Mode[/bold]")
-            console.print(f"[dim]Description: {agent_spec.description}[/dim]")
-            console.print(f"[dim]Max steps: {agent_spec.max_steps} | Session: {agent.session_id}[/dim]")
+            console.print(f"[dim]Description: {agent.agent_spec.description}[/dim]")
+            console.print(f"[dim]Max steps: {agent.agent_spec.max_steps} | Mode: {agent.agent_spec.mode.value} | Session: {agent.session_id}[/dim]")
             console.print("[dim]Commands: 'exit' to quit, 'clear' to clear history, 'info' for session info[/dim]")
             console.print()
 
@@ -298,12 +364,15 @@ def main(
         raise typer.Exit(code=1) from None
 
 
-@app.command(help="Enhanced agent chat (alias for default command)")
 @app.command(help="Enhanced agent chat (alias for default command)")
 def agent(
     ctx: typer.Context,
     message: Annotated[str | None, typer.Option("--once", help="Send one message and exit")] = None,
     agent_file: Annotated[str | None, typer.Option("--agent", help="Path to agent specification file")] = None,
+    agent_type: Annotated[
+        str | None,
+        typer.Option("--agent-type", "-t", help="Built-in agent type: coder, explorer, planner, focused_coder"),
+    ] = None,
     work_dir: Annotated[
         Path | None,
         typer.Option(
@@ -312,6 +381,9 @@ def agent(
     ] = None,
     no_progress: Annotated[bool, typer.Option("--no-progress", help="Disable progress indicators")] = False,
     list_agents: Annotated[bool, typer.Option("--list", help="List available agent specifications")] = False,
+    list_agent_types: Annotated[
+        bool, typer.Option("--list-types", help="List available built-in agent types")
+    ] = False,
     session_id: Annotated[
         str | None, typer.Option("--session", help="Use specific session ID for conversation continuity")
     ] = None,
@@ -325,9 +397,11 @@ def agent(
         main,
         message=message,
         agent_file=agent_file,
+        agent_type=agent_type,
         work_dir=work_dir,
         no_progress=no_progress,
         list_agents=list_agents,
+        list_agent_types=list_agent_types,
         session_id=session_id,
         clear_session=clear_session,
         list_sessions=list_sessions,

src/amcp/config.py

@@ -43,6 +43,12 @@ class ChatConfig:
     # Built-in file modification tools
     write_tool_enabled: bool | None = None
     edit_tool_enabled: bool | None = None
+    # Agent settings
+    default_agent: str | None = None  # default agent to use: "coder", "explorer", etc.
+    # Queue settings
+    enable_queue: bool | None = None  # enable message queue (default: True)
+    max_queue_size: int | None = None  # max queued messages per session (default: 100)
+
 
 
 @dataclass
@@ -97,6 +103,11 @@ def _decode_chat(raw: Mapping[str, object] | None) -> ChatConfig | None:
     mcp_servers = raw.get("mcp_servers")
     write_tool_enabled = raw.get("write_tool_enabled")
     edit_tool_enabled = raw.get("edit_tool_enabled")
+    # Agent settings
+    default_agent = raw.get("default_agent")
+    # Queue settings
+    enable_queue = raw.get("enable_queue")
+    max_queue_size = raw.get("max_queue_size")
     return ChatConfig(
         base_url=str(base_url) if base_url is not None else None,
         model=str(model) if model is not None else None,
@@ -109,6 +120,9 @@ def _decode_chat(raw: Mapping[str, object] | None) -> ChatConfig | None:
         mcp_servers=[str(s) for s in (mcp_servers or [])] if mcp_servers is not None else None,
         write_tool_enabled=bool(write_tool_enabled) if write_tool_enabled is not None else None,
         edit_tool_enabled=bool(edit_tool_enabled) if edit_tool_enabled is not None else None,
+        default_agent=str(default_agent) if default_agent is not None else None,
+        enable_queue=bool(enable_queue) if enable_queue is not None else None,
+        max_queue_size=int(max_queue_size) if max_queue_size is not None else None,
     )
 
 
@@ -160,6 +174,14 @@ def _encode_chat(c: ChatConfig | None) -> dict | None:
         out["write_tool_enabled"] = bool(c.write_tool_enabled)
     if c.edit_tool_enabled is not None:
         out["edit_tool_enabled"] = bool(c.edit_tool_enabled)
+    # Agent settings
+    if c.default_agent:
+        out["default_agent"] = c.default_agent
+    # Queue settings
+    if c.enable_queue is not None:
+        out["enable_queue"] = bool(c.enable_queue)
+    if c.max_queue_size is not None:
+        out["max_queue_size"] = int(c.max_queue_size)
     return out
 
 

src/amcp/message_queue.py

@@ -0,0 +1,531 @@
+"""
+Session-level Message Queue for AMCP.
+
+This module provides message queuing capabilities for agent sessions,
+inspired by Crush's sessionAgent message queue implementation.
+
+Key Features:
+- Session-specific message queues (messages are isolated per session)
+- Priority support for urgent messages
+- Automatic queuing when session is busy
+- Queue management (clear, peek, list)
+- Async/await compatible
+
+The queue ensures that:
+1. A session can only process one request at a time
+2. Incoming messages are queued while the session is busy
+3. Queued messages are processed in order when the session becomes free
+"""
+
+from __future__ import annotations
+
+import asyncio
+import uuid
+from collections import deque
+from dataclasses import dataclass, field
+from datetime import datetime
+from enum import Enum
+from typing import Any
+
+
+class MessagePriority(Enum):
+    """Priority levels for queued messages."""
+
+    LOW = 0
+    NORMAL = 1
+    HIGH = 2
+    URGENT = 3
+
+
+@dataclass
+class QueuedMessage:
+    """A message waiting in the queue.
+
+    Attributes:
+        id: Unique message identifier
+        session_id: Session this message belongs to
+        prompt: User's input prompt
+        attachments: Optional file attachments
+        priority: Message priority (default NORMAL)
+        created_at: When the message was queued
+        metadata: Additional message metadata
+    """
+
+    id: str
+    session_id: str
+    prompt: str
+    attachments: list[dict[str, Any]] = field(default_factory=list)
+    priority: MessagePriority = MessagePriority.NORMAL
+    created_at: datetime = field(default_factory=datetime.now)
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+    @classmethod
+    def create(
+        cls,
+        session_id: str,
+        prompt: str,
+        attachments: list[dict[str, Any]] | None = None,
+        priority: MessagePriority = MessagePriority.NORMAL,
+        **metadata: Any,
+    ) -> "QueuedMessage":
+        """Create a new queued message.
+
+        Args:
+            session_id: Target session ID
+            prompt: User's prompt text
+            attachments: Optional attachments
+            priority: Message priority
+            **metadata: Additional metadata
+
+        Returns:
+            New QueuedMessage instance
+        """
+        return cls(
+            id=str(uuid.uuid4()),
+            session_id=session_id,
+            prompt=prompt,
+            attachments=attachments or [],
+            priority=priority,
+            metadata=metadata,
+        )
+
+
+class SessionQueue:
+    """Message queue for a single session.
+
+    Manages a FIFO queue of messages for one session,
+    with support for priorities.
+    """
+
+    def __init__(self, session_id: str):
+        """Initialize session queue.
+
+        Args:
+            session_id: Session identifier
+        """
+        self.session_id = session_id
+        self._queue: deque[QueuedMessage] = deque()
+        self._lock = asyncio.Lock()
+
+    def __len__(self) -> int:
+        """Get the number of messages in the queue."""
+        return len(self._queue)
+
+    def is_empty(self) -> bool:
+        """Check if the queue is empty."""
+        return len(self._queue) == 0
+
+    async def enqueue(self, message: QueuedMessage) -> None:
+        """Add a message to the queue.
+
+        Messages are ordered by priority, then by creation time.
+
+        Args:
+            message: Message to add
+        """
+        async with self._lock:
+            # Find insertion point based on priority
+            insert_idx = len(self._queue)
+            for i, existing in enumerate(self._queue):
+                if message.priority.value > existing.priority.value:
+                    insert_idx = i
+                    break
+
+            self._queue.insert(insert_idx, message)
+
+    async def dequeue(self) -> QueuedMessage | None:
+        """Remove and return the next message from the queue.
+
+        Returns:
+            Next message or None if queue is empty
+        """
+        async with self._lock:
+            if self._queue:
+                return self._queue.popleft()
+            return None
+
+    async def peek(self) -> QueuedMessage | None:
+        """Look at the next message without removing it.
+
+        Returns:
+            Next message or None if queue is empty
+        """
+        async with self._lock:
+            if self._queue:
+                return self._queue[0]
+            return None
+
+    async def clear(self) -> int:
+        """Clear all messages from the queue.
+
+        Returns:
+            Number of messages cleared
+        """
+        async with self._lock:
+            count = len(self._queue)
+            self._queue.clear()
+            return count
+
+    def list_messages(self) -> list[QueuedMessage]:
+        """Get a copy of all queued messages.
+
+        Returns:
+            List of queued messages
+        """
+        return list(self._queue)
+
+    def list_prompts(self) -> list[str]:
+        """Get list of queued prompts for display.
+
+        Returns:
+            List of prompt strings
+        """
+        return [msg.prompt for msg in self._queue]
+
+
+class MessageQueueManager:
+    """Manages message queues for multiple sessions.
+
+    Provides centralized management of session queues,
+    busy state tracking, and message routing.
+
+    Example usage:
+        manager = MessageQueueManager()
+
+        # Check if session is busy
+        if manager.is_busy("session-123"):
+            # Enqueue the message
+            await manager.enqueue("session-123", "Hello!")
+        else:
+            # Process immediately
+            await manager.acquire("session-123")
+            try:
+                result = await process_message(prompt)
+            finally:
+                manager.release("session-123")
+
+            # Process any queued messages
+            while True:
+                next_msg = await manager.dequeue("session-123")
+                if not next_msg:
+                    break
+                # Process next_msg...
+    """
+
+    def __init__(self):
+        """Initialize the message queue manager."""
+        self._session_queues: dict[str, SessionQueue] = {}
+        self._busy_sessions: set[str] = set()
+        self._session_locks: dict[str, asyncio.Lock] = {}
+        self._global_lock = asyncio.Lock()
+
+    def _get_or_create_queue(self, session_id: str) -> SessionQueue:
+        """Get or create a queue for a session.
+
+        Args:
+            session_id: Session identifier
+
+        Returns:
+            Session queue
+        """
+        if session_id not in self._session_queues:
+            self._session_queues[session_id] = SessionQueue(session_id)
+        return self._session_queues[session_id]
+
+    def _get_or_create_lock(self, session_id: str) -> asyncio.Lock:
+        """Get or create a lock for a session.
+
+        Args:
+            session_id: Session identifier
+
+        Returns:
+            Session lock
+        """
+        if session_id not in self._session_locks:
+            self._session_locks[session_id] = asyncio.Lock()
+        return self._session_locks[session_id]
+
+    def is_busy(self, session_id: str) -> bool:
+        """Check if a session is currently busy.
+
+        Args:
+            session_id: Session identifier
+
+        Returns:
+            True if session is processing a request
+        """
+        return session_id in self._busy_sessions
+
+    def any_busy(self) -> bool:
+        """Check if any session is busy.
+
+        Returns:
+            True if any session is processing
+        """
+        return len(self._busy_sessions) > 0
+
+    def get_busy_sessions(self) -> list[str]:
+        """Get list of busy session IDs.
+
+        Returns:
+            List of busy session IDs
+        """
+        return list(self._busy_sessions)
+
+    async def acquire(self, session_id: str) -> bool:
+        """Acquire exclusive access to a session.
+
+        Use this before processing a request for a session.
+
+        Args:
+            session_id: Session identifier
+
+        Returns:
+            True if acquired, False if session was already busy
+        """
+        async with self._global_lock:
+            if session_id in self._busy_sessions:
+                return False
+            self._busy_sessions.add(session_id)
+            return True
+
+    def release(self, session_id: str) -> None:
+        """Release a session after processing.
+
+        Use this after finishing processing a request.
+
+        Args:
+            session_id: Session identifier
+        """
+        self._busy_sessions.discard(session_id)
+
+    async def enqueue(
+        self,
+        session_id: str,
+        prompt: str,
+        attachments: list[dict[str, Any]] | None = None,
+        priority: MessagePriority = MessagePriority.NORMAL,
+        **metadata: Any,
+    ) -> QueuedMessage:
+        """Add a message to a session's queue.
+
+        Use this when a session is busy and cannot process immediately.
+
+        Args:
+            session_id: Target session ID
+            prompt: User's prompt text
+            attachments: Optional attachments
+            priority: Message priority
+            **metadata: Additional metadata
+
+        Returns:
+            The queued message
+        """
+        queue = self._get_or_create_queue(session_id)
+        message = QueuedMessage.create(
+            session_id=session_id,
+            prompt=prompt,
+            attachments=attachments,
+            priority=priority,
+            **metadata,
+        )
+        await queue.enqueue(message)
+        return message
+
+    async def enqueue_if_busy(
+        self,
+        session_id: str,
+        prompt: str,
+        attachments: list[dict[str, Any]] | None = None,
+        priority: MessagePriority = MessagePriority.NORMAL,
+        **metadata: Any,
+    ) -> tuple[bool, QueuedMessage | None]:
+        """Automatically enqueue a message if session is busy.
+
+        This is a convenience method that:
+        1. Checks if the session is busy
+        2. If busy, queues the message and returns (True, message)
+        3. If not busy, returns (False, None)
+
+        Args:
+            session_id: Target session ID
+            prompt: User's prompt text
+            attachments: Optional attachments
+            priority: Message priority
+            **metadata: Additional metadata
+
+        Returns:
+            Tuple of (was_queued, queued_message)
+        """
+        if self.is_busy(session_id):
+            message = await self.enqueue(session_id, prompt, attachments, priority, **metadata)
+            return (True, message)
+        return (False, None)
+
+    async def dequeue(self, session_id: str) -> QueuedMessage | None:
+        """Get the next queued message for a session.
+
+        Args:
+            session_id: Session identifier
+
+        Returns:
+            Next message or None if queue is empty
+        """
+        queue = self._get_or_create_queue(session_id)
+        return await queue.dequeue()
+
+    async def peek(self, session_id: str) -> QueuedMessage | None:
+        """Look at the next message without removing it.
+
+        Args:
+            session_id: Session identifier
+
+        Returns:
+            Next message or None if queue is empty
+        """
+        queue = self._get_or_create_queue(session_id)
+        return await queue.peek()
+
+    def queued_count(self, session_id: str) -> int:
+        """Get the number of queued messages for a session.
+
+        Args:
+            session_id: Session identifier
+
+        Returns:
+            Number of queued messages
+        """
+        if session_id not in self._session_queues:
+            return 0
+        return len(self._session_queues[session_id])
+
+    def queued_prompts(self, session_id: str) -> list[str]:
+        """Get list of queued prompts for a session.
+
+        Useful for displaying queue status to users.
+
+        Args:
+            session_id: Session identifier
+
+        Returns:
+            List of prompt strings
+        """
+        if session_id not in self._session_queues:
+            return []
+        return self._session_queues[session_id].list_prompts()
+
+    async def clear_queue(self, session_id: str) -> int:
+        """Clear all queued messages for a session.
+
+        Args:
+            session_id: Session identifier
+
+        Returns:
+            Number of messages cleared
+        """
+        if session_id not in self._session_queues:
+            return 0
+        return await self._session_queues[session_id].clear()
+
+    def get_queue_status(self, session_id: str) -> dict[str, Any]:
+        """Get detailed queue status for a session.
+
+        Args:
+            session_id: Session identifier
+
+        Returns:
+            Status dictionary
+        """
+        return {
+            "session_id": session_id,
+            "is_busy": self.is_busy(session_id),
+            "queued_count": self.queued_count(session_id),
+            "queued_prompts": self.queued_prompts(session_id),
+        }
+
+    def get_all_status(self) -> dict[str, Any]:
+        """Get status for all sessions.
+
+        Returns:
+            Status dictionary for all sessions
+        """
+        return {
+            "busy_sessions": self.get_busy_sessions(),
+            "total_queued": sum(len(q) for q in self._session_queues.values()),
+            "sessions": {
+                sid: self.get_queue_status(sid)
+                for sid in set(self._session_queues.keys()) | self._busy_sessions
+            },
+        }
+
+
+# Global message queue manager singleton
+_queue_manager: MessageQueueManager | None = None
+
+
+def get_message_queue_manager() -> MessageQueueManager:
+    """Get the global message queue manager.
+
+    Returns:
+        Global MessageQueueManager instance
+    """
+    global _queue_manager
+    if _queue_manager is None:
+        _queue_manager = MessageQueueManager()
+    return _queue_manager
+
+
+async def run_with_queue(
+    session_id: str,
+    prompt: str,
+    processor_fn,
+    attachments: list[dict[str, Any]] | None = None,
+    priority: MessagePriority = MessagePriority.NORMAL,
+) -> Any | None:
+    """Run a prompt through the queue system.
+
+    This is a high-level helper that:
+    1. Queues the message if session is busy
+    2. Otherwise processes immediately
+    3. After processing, checks for and processes queued messages
+
+    Args:
+        session_id: Session identifier
+        prompt: User's prompt
+        processor_fn: Async function(prompt, attachments) to process the message
+        attachments: Optional attachments
+        priority: Message priority
+
+    Returns:
+        Result from processor_fn, or None if queued
+    """
+    manager = get_message_queue_manager()
+
+    # Check if we should queue
+    was_queued, _ = await manager.enqueue_if_busy(session_id, prompt, attachments, priority)
+    if was_queued:
+        return None
+
+    # Acquire the session
+    acquired = await manager.acquire(session_id)
+    if not acquired:
+        # Race condition - queue it
+        await manager.enqueue(session_id, prompt, attachments, priority)
+        return None
+
+    try:
+        # Process the message
+        result = await processor_fn(prompt, attachments or [])
+
+        # Process any queued messages
+        while True:
+            next_msg = await manager.dequeue(session_id)
+            if not next_msg:
+                break
+            # Process queued message
+            await processor_fn(next_msg.prompt, next_msg.attachments)
+
+        return result
+
+    finally:
+        manager.release(session_id)

src/amcp/multi_agent.py

@@ -0,0 +1,375 @@
+"""
+Multi-Agent System for AMCP.
+
+This module provides support for multiple agents with different capabilities,
+including Primary agents and Subagents with explicit mode differentiation.
+
+Inspired by:
+- OpenCode's agent modes (primary, subagent, explore, plan)
+- Crush's Coordinator pattern with message queuing
+- Kimi-CLI's LaborMarket for agent management
+
+Features:
+- AgentMode.PRIMARY / AgentMode.SUBAGENT differentiation
+- Built-in agent configurations for common tasks
+- Agent registry for dynamic agent lookup
+- Support for agent delegation
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from enum import Enum
+from pathlib import Path
+from typing import Any
+
+
+class AgentMode(Enum):
+    """Agent execution mode."""
+
+    PRIMARY = "primary"
+    """Primary agent with full capabilities - can delegate to subagents."""
+
+    SUBAGENT = "subagent"
+    """Subagent with restricted capabilities - focuses on single tasks."""
+
+
+@dataclass
+class AgentConfig:
+    """Configuration for an agent type.
+
+    Attributes:
+        name: Unique identifier for the agent
+        mode: PRIMARY for main agents, SUBAGENT for task-specific agents
+        description: Human-readable description of agent's purpose
+        system_prompt: System prompt template for the agent
+        tools: List of tool names the agent can use
+        excluded_tools: Tools explicitly disabled for this agent
+        max_steps: Maximum execution steps for this agent
+        can_delegate: Whether this agent can spawn subagents
+        parent_agent: Name of parent agent (for subagents)
+    """
+
+    name: str
+    mode: AgentMode
+    description: str
+    system_prompt: str
+    tools: list[str] = field(default_factory=list)
+    excluded_tools: list[str] = field(default_factory=list)
+    max_steps: int = 100
+    can_delegate: bool = True
+    parent_agent: str | None = None
+
+    def __post_init__(self):
+        """Validate agent configuration."""
+        # Subagents cannot delegate by default
+        if self.mode == AgentMode.SUBAGENT:
+            self.can_delegate = False
+
+    def get_effective_tools(self, available_tools: list[str]) -> list[str]:
+        """Get the effective list of tools for this agent.
+
+        Args:
+            available_tools: All available tools in the system
+
+        Returns:
+            List of tool names this agent can use
+        """
+        if self.tools:
+            # Use explicit whitelist
+            effective = [t for t in self.tools if t in available_tools]
+        else:
+            # Start with all available tools
+            effective = list(available_tools)
+
+        # Apply exclusions
+        return [t for t in effective if t not in self.excluded_tools]
+
+
+# Default system prompt templates
+PRIMARY_SYSTEM_PROMPT = """You are {agent_name}, an AI coding assistant with full capabilities.
+
+You can use all available tools to help users with software engineering tasks.
+When tasks are complex, you may delegate to specialized subagents.
+
+Current working directory: {work_dir}
+Current time: {current_time}
+
+Guidelines:
+- Use appropriate tools for each task
+- Read files to understand the codebase before making changes
+- Delegate complex sub-tasks to specialized agents when needed
+- Be precise and efficient in your tool usage
+- Explain your actions when helpful"""
+
+EXPLORER_SYSTEM_PROMPT = """You are {agent_name}, a fast codebase exploration agent.
+
+Your task is to quickly analyze and understand codebases WITHOUT making changes.
+You have READ-ONLY access to files and search tools.
+
+Current working directory: {work_dir}
+Current time: {current_time}
+
+Guidelines:
+- Focus on quick exploration and understanding
+- Do NOT attempt to modify any files
+- Summarize your findings concisely
+- Report back to the main agent when done"""
+
+PLANNER_SYSTEM_PROMPT = """You are {agent_name}, a planning and analysis agent.
+
+Your task is to analyze problems and create execution plans WITHOUT implementing them.
+You have READ-ONLY access to the codebase.
+
+Current working directory: {work_dir}
+Current time: {current_time}
+
+Guidelines:
+- Create detailed, step-by-step plans
+- Identify potential issues and edge cases
+- Do NOT implement the plan yourself
+- Return a clear, actionable plan to the main agent"""
+
+CODER_SYSTEM_PROMPT = """You are {agent_name}, a focused coding agent.
+
+Your task is to implement specific code changes as directed.
+You have full write access to the codebase.
+
+Current working directory: {work_dir}
+Current time: {current_time}
+
+Guidelines:
+- Implement exactly what is requested
+- Follow existing code patterns and style
+- Test your changes when possible
+- Keep changes minimal and focused"""
+
+# Built-in agent configurations
+BUILTIN_AGENTS: dict[str, AgentConfig] = {
+    "coder": AgentConfig(
+        name="coder",
+        mode=AgentMode.PRIMARY,
+        description="Main coding agent with full capabilities",
+        system_prompt=PRIMARY_SYSTEM_PROMPT,
+        tools=[],  # Empty means all available tools
+        excluded_tools=[],
+        max_steps=300,
+        can_delegate=True,
+    ),
+    "explorer": AgentConfig(
+        name="explorer",
+        mode=AgentMode.SUBAGENT,
+        description="Fast codebase exploration agent (read-only)",
+        system_prompt=EXPLORER_SYSTEM_PROMPT,
+        tools=["read_file", "grep", "glob", "think"],
+        excluded_tools=["write_file", "edit_file", "bash"],
+        max_steps=50,
+        can_delegate=False,
+    ),
+    "planner": AgentConfig(
+        name="planner",
+        mode=AgentMode.SUBAGENT,
+        description="Planning agent for analysis and strategy (read-only)",
+        system_prompt=PLANNER_SYSTEM_PROMPT,
+        tools=["read_file", "grep", "glob", "think"],
+        excluded_tools=["write_file", "edit_file", "bash"],
+        max_steps=30,
+        can_delegate=False,
+    ),
+    "focused_coder": AgentConfig(
+        name="focused_coder",
+        mode=AgentMode.SUBAGENT,
+        description="Focused coding agent for specific implementation tasks",
+        system_prompt=CODER_SYSTEM_PROMPT,
+        tools=["read_file", "grep", "write_file", "edit_file", "bash", "think"],
+        excluded_tools=[],
+        max_steps=100,
+        can_delegate=False,
+    ),
+}
+
+
+class AgentRegistry:
+    """Registry for managing and looking up agent configurations.
+
+    This class provides centralized management of agent configurations,
+    supporting both built-in and custom agents.
+    """
+
+    def __init__(self):
+        """Initialize the agent registry with built-in agents."""
+        self._agents: dict[str, AgentConfig] = dict(BUILTIN_AGENTS)
+        self._custom_agents: dict[str, AgentConfig] = {}
+
+    def register(self, config: AgentConfig) -> None:
+        """Register a custom agent configuration.
+
+        Args:
+            config: Agent configuration to register
+        """
+        self._custom_agents[config.name] = config
+        self._agents[config.name] = config
+
+    def get(self, name: str) -> AgentConfig | None:
+        """Get an agent configuration by name.
+
+        Args:
+            name: Agent name
+
+        Returns:
+            Agent configuration or None if not found
+        """
+        return self._agents.get(name)
+
+    def list_agents(self) -> list[str]:
+        """List all registered agent names.
+
+        Returns:
+            List of agent names
+        """
+        return list(self._agents.keys())
+
+    def list_primary_agents(self) -> list[str]:
+        """List all primary agent names.
+
+        Returns:
+            List of primary agent names
+        """
+        return [name for name, cfg in self._agents.items() if cfg.mode == AgentMode.PRIMARY]
+
+    def list_subagents(self) -> list[str]:
+        """List all subagent names.
+
+        Returns:
+            List of subagent names
+        """
+        return [name for name, cfg in self._agents.items() if cfg.mode == AgentMode.SUBAGENT]
+
+    def get_subagents_for(self, parent_name: str) -> list[str]:
+        """Get subagents that can be used by a parent agent.
+
+        Args:
+            parent_name: Name of the parent agent
+
+        Returns:
+            List of subagent names available to the parent
+        """
+        parent = self.get(parent_name)
+        if not parent or not parent.can_delegate:
+            return []
+
+        return self.list_subagents()
+
+    def load_from_file(self, config_file: Path) -> None:
+        """Load agent configurations from a YAML file.
+
+        Args:
+            config_file: Path to YAML configuration file
+        """
+        import yaml
+
+        if not config_file.exists():
+            return
+
+        try:
+            with open(config_file, encoding="utf-8") as f:
+                data = yaml.safe_load(f)
+
+            if not data or "agents" not in data:
+                return
+
+            for agent_data in data.get("agents", []):
+                # Parse mode
+                mode_str = agent_data.get("mode", "primary")
+                mode = AgentMode.PRIMARY if mode_str == "primary" else AgentMode.SUBAGENT
+
+                config = AgentConfig(
+                    name=agent_data.get("name", "custom"),
+                    mode=mode,
+                    description=agent_data.get("description", ""),
+                    system_prompt=agent_data.get("system_prompt", PRIMARY_SYSTEM_PROMPT),
+                    tools=agent_data.get("tools", []),
+                    excluded_tools=agent_data.get("excluded_tools", []),
+                    max_steps=agent_data.get("max_steps", 100),
+                    can_delegate=agent_data.get("can_delegate", mode == AgentMode.PRIMARY),
+                    parent_agent=agent_data.get("parent_agent"),
+                )
+                self.register(config)
+
+        except Exception as e:
+            # Log but don't fail
+            print(f"Warning: Could not load agent config file: {e}")
+
+
+# Global agent registry singleton
+_agent_registry: AgentRegistry | None = None
+
+
+def get_agent_registry() -> AgentRegistry:
+    """Get the global agent registry.
+
+    Returns:
+        Global AgentRegistry instance
+    """
+    global _agent_registry
+    if _agent_registry is None:
+        _agent_registry = AgentRegistry()
+    return _agent_registry
+
+
+def get_agent_config(name: str) -> AgentConfig | None:
+    """Get an agent configuration by name.
+
+    Args:
+        name: Agent name
+
+    Returns:
+        Agent configuration or None if not found
+    """
+    return get_agent_registry().get(name)
+
+
+def create_subagent_config(
+    parent_name: str,
+    task_description: str,
+    tools: list[str] | None = None,
+) -> AgentConfig:
+    """Create a dynamic subagent configuration for a specific task.
+
+    This is used for creating task-specific agents at runtime.
+
+    Args:
+        parent_name: Name of the parent agent
+        task_description: Description of the task for the subagent
+        tools: Optional list of tools for the subagent
+
+    Returns:
+        New AgentConfig for the subagent
+    """
+    import uuid
+
+    subagent_id = str(uuid.uuid4())[:8]
+    subagent_name = f"task_{subagent_id}"
+
+    system_prompt = f"""You are a specialized task agent for: {task_description}
+
+Complete this specific task and report back when done.
+Be focused and efficient in completing the assigned task.
+
+Current working directory: {{work_dir}}
+Current time: {{current_time}}
+
+Task: {task_description}
+"""
+
+    return AgentConfig(
+        name=subagent_name,
+        mode=AgentMode.SUBAGENT,
+        description=task_description,
+        system_prompt=system_prompt,
+        tools=tools or ["read_file", "grep", "write_file", "edit_file", "think"],
+        excluded_tools=[],
+        max_steps=50,
+        can_delegate=False,
+        parent_agent=parent_name,
+    )

tests/test_agent_spec.py

@@ -1,7 +1,247 @@
-from amcp.agent_spec import get_default_agent_spec
+"""Tests for the agent_spec module."""
 
+import pytest
+from pathlib import Path
 
-def test_agent_spec_default():
-    spec = get_default_agent_spec()
-    assert spec.name == "default"
-    assert len(spec.system_prompt) > 0
+from amcp.agent_spec import (
+    AgentSpec,
+    AgentSpecError,
+    ResolvedAgentSpec,
+    get_default_agent_spec,
+    load_agent_spec,
+    list_available_agents,
+)
+from amcp.multi_agent import AgentMode
+
+
+class TestGetDefaultAgentSpec:
+    """Tests for get_default_agent_spec function."""
+
+    def test_returns_resolved_spec(self):
+        """Test that get_default_agent_spec returns a ResolvedAgentSpec."""
+        spec = get_default_agent_spec()
+        assert isinstance(spec, ResolvedAgentSpec)
+
+    def test_default_name(self):
+        """Test default agent name."""
+        spec = get_default_agent_spec()
+        assert spec.name == "default"
+
+    def test_default_description(self):
+        """Test default agent description."""
+        spec = get_default_agent_spec()
+        assert spec.description == "Default AMCP agent"
+
+    def test_default_mode_is_primary(self):
+        """Test that default agent is PRIMARY mode."""
+        spec = get_default_agent_spec()
+        assert spec.mode == AgentMode.PRIMARY
+
+    def test_default_can_delegate(self):
+        """Test that default agent can delegate."""
+        spec = get_default_agent_spec()
+        assert spec.can_delegate is True
+
+    def test_has_system_prompt(self):
+        """Test that default agent has a system prompt."""
+        spec = get_default_agent_spec()
+        assert len(spec.system_prompt) > 0
+        assert "{work_dir}" in spec.system_prompt
+        assert "{current_time}" in spec.system_prompt
+
+    def test_default_max_steps(self):
+        """Test default max_steps value."""
+        spec = get_default_agent_spec()
+        assert spec.max_steps == 300
+
+
+class TestAgentSpec:
+    """Tests for AgentSpec Pydantic model."""
+
+    def test_minimal_spec(self):
+        """Test creating a minimal agent spec."""
+        spec = AgentSpec(
+            name="test",
+            system_prompt="You are a test agent.",
+        )
+        assert spec.name == "test"
+        assert spec.mode == "primary"  # Default
+        assert spec.can_delegate is True  # Default
+
+    def test_spec_with_mode(self):
+        """Test creating spec with explicit mode."""
+        spec = AgentSpec(
+            name="test",
+            system_prompt="Test",
+            mode="subagent",
+        )
+        assert spec.mode == "subagent"
+
+    def test_spec_with_can_delegate(self):
+        """Test creating spec with can_delegate."""
+        spec = AgentSpec(
+            name="test",
+            system_prompt="Test",
+            can_delegate=False,
+        )
+        assert spec.can_delegate is False
+
+    def test_default_values(self):
+        """Test default values."""
+        spec = AgentSpec(
+            name="test",
+            system_prompt="Test",
+        )
+        assert spec.description == ""
+        assert spec.tools == []
+        assert spec.exclude_tools == []
+        assert spec.max_steps == 5
+        assert spec.model == ""
+        assert spec.base_url == ""
+
+
+class TestLoadAgentSpec:
+    """Tests for load_agent_spec function."""
+
+    def test_load_nonexistent_file(self, tmp_path):
+        """Test loading a non-existent file raises error."""
+        with pytest.raises(AgentSpecError, match="not found"):
+            load_agent_spec(tmp_path / "nonexistent.yaml")
+
+    def test_load_empty_file(self, tmp_path):
+        """Test loading an empty file raises error."""
+        empty_file = tmp_path / "empty.yaml"
+        empty_file.write_text("")
+
+        with pytest.raises(AgentSpecError, match="Empty"):
+            load_agent_spec(empty_file)
+
+    def test_load_invalid_yaml(self, tmp_path):
+        """Test loading invalid YAML raises error."""
+        invalid_file = tmp_path / "invalid.yaml"
+        invalid_file.write_text("name: [invalid")
+
+        with pytest.raises(AgentSpecError, match="Invalid YAML"):
+            load_agent_spec(invalid_file)
+
+    def test_load_valid_spec(self, tmp_path):
+        """Test loading a valid agent spec."""
+        spec_file = tmp_path / "agent.yaml"
+        spec_file.write_text("""
+name: test_agent
+description: A test agent
+mode: primary
+system_prompt: You are a test agent.
+tools:
+  - read_file
+  - grep
+max_steps: 50
+can_delegate: true
+""")
+        spec = load_agent_spec(spec_file)
+        assert spec.name == "test_agent"
+        assert spec.description == "A test agent"
+        assert spec.mode == AgentMode.PRIMARY
+        assert "read_file" in spec.tools
+        assert spec.max_steps == 50
+        assert spec.can_delegate is True
+
+    def test_load_subagent_spec(self, tmp_path):
+        """Test loading a subagent spec sets can_delegate to False."""
+        spec_file = tmp_path / "subagent.yaml"
+        spec_file.write_text("""
+name: test_subagent
+mode: subagent
+system_prompt: You are a test subagent.
+can_delegate: true  # This should be overridden
+""")
+        spec = load_agent_spec(spec_file)
+        assert spec.mode == AgentMode.SUBAGENT
+        assert spec.can_delegate is False  # Should be overridden
+
+    def test_load_spec_with_template(self, tmp_path):
+        """Test loading spec with system prompt template."""
+        spec_file = tmp_path / "template.yaml"
+        spec_file.write_text("""
+name: template_agent
+system_prompt: ""
+system_prompt_template: "Hello, {name}! You work in {location}."
+system_prompt_vars:
+  name: TestBot
+  location: TestLand
+""")
+        spec = load_agent_spec(spec_file)
+        assert "Hello, TestBot!" in spec.system_prompt
+        assert "TestLand" in spec.system_prompt
+
+
+class TestListAvailableAgents:
+    """Tests for list_available_agents function."""
+
+    def test_empty_directory(self, tmp_path):
+        """Test listing agents in empty directory."""
+        agents = list_available_agents(tmp_path)
+        assert agents == []
+
+    def test_nonexistent_directory(self, tmp_path):
+        """Test listing agents in non-existent directory."""
+        agents = list_available_agents(tmp_path / "nonexistent")
+        assert agents == []
+
+    def test_finds_yaml_files(self, tmp_path):
+        """Test that YAML files are found."""
+        (tmp_path / "agent1.yaml").write_text("name: agent1\nsystem_prompt: test")
+        (tmp_path / "agent2.yaml").write_text("name: agent2\nsystem_prompt: test")
+        (tmp_path / "not_yaml.txt").write_text("not yaml")
+
+        agents = list_available_agents(tmp_path)
+        assert len(agents) == 2
+        assert any(a.name == "agent1.yaml" for a in agents)
+        assert any(a.name == "agent2.yaml" for a in agents)
+
+    def test_finds_nested_yaml_files(self, tmp_path):
+        """Test that nested YAML files are found."""
+        subdir = tmp_path / "subdir"
+        subdir.mkdir()
+        (subdir / "nested.yaml").write_text("name: nested\nsystem_prompt: test")
+
+        agents = list_available_agents(tmp_path)
+        assert len(agents) == 1
+        assert "nested.yaml" in str(agents[0])
+
+
+class TestResolvedAgentSpec:
+    """Tests for ResolvedAgentSpec dataclass."""
+
+    def test_all_fields(self):
+        """Test creating with all fields."""
+        spec = ResolvedAgentSpec(
+            name="test",
+            description="Test description",
+            mode=AgentMode.PRIMARY,
+            system_prompt="Test prompt",
+            tools=["read_file"],
+            exclude_tools=["bash"],
+            max_steps=100,
+            model="test-model",
+            base_url="https://test.api",
+            can_delegate=True,
+        )
+        assert spec.name == "test"
+        assert spec.mode == AgentMode.PRIMARY
+        assert spec.can_delegate is True
+
+    def test_default_can_delegate(self):
+        """Test that can_delegate defaults to True."""
+        spec = ResolvedAgentSpec(
+            name="test",
+            description="",
+            mode=AgentMode.PRIMARY,
+            system_prompt="",
+            tools=[],
+            exclude_tools=[],
+            max_steps=100,
+            model="",
+            base_url="",
+        )
+        assert spec.can_delegate is True

tests/test_message_queue.py

@@ -0,0 +1,372 @@
+"""Tests for the message_queue module."""
+
+import asyncio
+
+import pytest
+
+from amcp.message_queue import (
+    MessagePriority,
+    MessageQueueManager,
+    QueuedMessage,
+    SessionQueue,
+    get_message_queue_manager,
+    run_with_queue,
+)
+
+
+class TestMessagePriority:
+    """Tests for MessagePriority enum."""
+
+    def test_priority_values(self):
+        """Test that priority values are ordered correctly."""
+        assert MessagePriority.LOW.value < MessagePriority.NORMAL.value
+        assert MessagePriority.NORMAL.value < MessagePriority.HIGH.value
+        assert MessagePriority.HIGH.value < MessagePriority.URGENT.value
+
+    def test_priority_comparison(self):
+        """Test priority comparison."""
+        assert MessagePriority.URGENT.value > MessagePriority.LOW.value
+        assert MessagePriority.NORMAL.value == 1
+
+
+class TestQueuedMessage:
+    """Tests for QueuedMessage dataclass."""
+
+    def test_create_with_defaults(self):
+        """Test creating a message with default values."""
+        msg = QueuedMessage.create(
+            session_id="test-session",
+            prompt="Hello, world!",
+        )
+        assert msg.session_id == "test-session"
+        assert msg.prompt == "Hello, world!"
+        assert msg.priority == MessagePriority.NORMAL
+        assert msg.attachments == []
+        assert msg.metadata == {}
+        assert msg.id is not None
+        assert msg.created_at is not None
+
+    def test_create_with_priority(self):
+        """Test creating a message with custom priority."""
+        msg = QueuedMessage.create(
+            session_id="test-session",
+            prompt="Urgent!",
+            priority=MessagePriority.URGENT,
+        )
+        assert msg.priority == MessagePriority.URGENT
+
+    def test_create_with_attachments(self):
+        """Test creating a message with attachments."""
+        attachments = [{"file": "test.txt", "content": "Hello"}]
+        msg = QueuedMessage.create(
+            session_id="test-session",
+            prompt="With attachment",
+            attachments=attachments,
+        )
+        assert msg.attachments == attachments
+
+    def test_create_with_metadata(self):
+        """Test creating a message with metadata."""
+        msg = QueuedMessage.create(
+            session_id="test-session",
+            prompt="Test",
+            work_dir="/home/user",
+            stream=True,
+        )
+        assert msg.metadata["work_dir"] == "/home/user"
+        assert msg.metadata["stream"] is True
+
+    def test_unique_ids(self):
+        """Test that each message gets a unique ID."""
+        msg1 = QueuedMessage.create("session", "Prompt 1")
+        msg2 = QueuedMessage.create("session", "Prompt 2")
+        assert msg1.id != msg2.id
+
+
+class TestSessionQueue:
+    """Tests for SessionQueue class."""
+
+    @pytest.fixture
+    def queue(self):
+        """Create a fresh queue for each test."""
+        return SessionQueue("test-session")
+
+    @pytest.mark.asyncio
+    async def test_enqueue_dequeue(self, queue):
+        """Test basic enqueue and dequeue."""
+        msg = QueuedMessage.create("test-session", "Hello")
+        await queue.enqueue(msg)
+        assert len(queue) == 1
+
+        dequeued = await queue.dequeue()
+        assert dequeued is not None
+        assert dequeued.prompt == "Hello"
+        assert len(queue) == 0
+
+    @pytest.mark.asyncio
+    async def test_is_empty(self, queue):
+        """Test is_empty method."""
+        assert queue.is_empty()
+        await queue.enqueue(QueuedMessage.create("test-session", "Hello"))
+        assert not queue.is_empty()
+
+    @pytest.mark.asyncio
+    async def test_priority_ordering(self, queue):
+        """Test that messages are ordered by priority."""
+        await queue.enqueue(QueuedMessage.create("s", "Normal", priority=MessagePriority.NORMAL))
+        await queue.enqueue(QueuedMessage.create("s", "Low", priority=MessagePriority.LOW))
+        await queue.enqueue(QueuedMessage.create("s", "Urgent", priority=MessagePriority.URGENT))
+        await queue.enqueue(QueuedMessage.create("s", "High", priority=MessagePriority.HIGH))
+
+        # Should dequeue in priority order: URGENT, HIGH, NORMAL, LOW
+        prompts = []
+        while not queue.is_empty():
+            msg = await queue.dequeue()
+            prompts.append(msg.prompt)
+
+        assert prompts == ["Urgent", "High", "Normal", "Low"]
+
+    @pytest.mark.asyncio
+    async def test_peek(self, queue):
+        """Test peek without removing."""
+        msg = QueuedMessage.create("test-session", "Hello")
+        await queue.enqueue(msg)
+
+        peeked = await queue.peek()
+        assert peeked is not None
+        assert peeked.prompt == "Hello"
+        assert len(queue) == 1  # Still in queue
+
+    @pytest.mark.asyncio
+    async def test_dequeue_empty(self, queue):
+        """Test dequeue from empty queue returns None."""
+        result = await queue.dequeue()
+        assert result is None
+
+    @pytest.mark.asyncio
+    async def test_clear(self, queue):
+        """Test clearing the queue."""
+        await queue.enqueue(QueuedMessage.create("s", "1"))
+        await queue.enqueue(QueuedMessage.create("s", "2"))
+        await queue.enqueue(QueuedMessage.create("s", "3"))
+
+        count = await queue.clear()
+        assert count == 3
+        assert queue.is_empty()
+
+    def test_list_messages(self, queue):
+        """Test listing messages without async."""
+        # We need to run async operations
+        asyncio.run(queue.enqueue(QueuedMessage.create("s", "Hello")))
+        messages = queue.list_messages()
+        assert len(messages) == 1
+        assert messages[0].prompt == "Hello"
+
+    def test_list_prompts(self, queue):
+        """Test listing prompts."""
+        asyncio.run(queue.enqueue(QueuedMessage.create("s", "Hello")))
+        asyncio.run(queue.enqueue(QueuedMessage.create("s", "World")))
+        prompts = queue.list_prompts()
+        assert prompts == ["Hello", "World"]
+
+
+class TestMessageQueueManager:
+    """Tests for MessageQueueManager class."""
+
+    @pytest.fixture
+    def manager(self):
+        """Create a fresh manager for each test."""
+        return MessageQueueManager()
+
+    @pytest.mark.asyncio
+    async def test_acquire_release(self, manager):
+        """Test session acquire and release."""
+        assert not manager.is_busy("session-1")
+
+        acquired = await manager.acquire("session-1")
+        assert acquired
+        assert manager.is_busy("session-1")
+
+        manager.release("session-1")
+        assert not manager.is_busy("session-1")
+
+    @pytest.mark.asyncio
+    async def test_acquire_fails_when_busy(self, manager):
+        """Test that acquire fails when session is already busy."""
+        await manager.acquire("session-1")
+        assert manager.is_busy("session-1")
+
+        # Try to acquire again
+        acquired_again = await manager.acquire("session-1")
+        assert not acquired_again
+
+    @pytest.mark.asyncio
+    async def test_multiple_sessions(self, manager):
+        """Test managing multiple sessions."""
+        await manager.acquire("session-1")
+        await manager.acquire("session-2")
+
+        assert manager.is_busy("session-1")
+        assert manager.is_busy("session-2")
+        assert not manager.is_busy("session-3")
+
+        busy_sessions = manager.get_busy_sessions()
+        assert "session-1" in busy_sessions
+        assert "session-2" in busy_sessions
+
+    @pytest.mark.asyncio
+    async def test_any_busy(self, manager):
+        """Test any_busy method."""
+        assert not manager.any_busy()
+
+        await manager.acquire("session-1")
+        assert manager.any_busy()
+
+        manager.release("session-1")
+        assert not manager.any_busy()
+
+    @pytest.mark.asyncio
+    async def test_enqueue(self, manager):
+        """Test enqueueing messages."""
+        msg = await manager.enqueue("session-1", "Hello")
+        assert msg.prompt == "Hello"
+        assert manager.queued_count("session-1") == 1
+
+    @pytest.mark.asyncio
+    async def test_enqueue_if_busy(self, manager):
+        """Test enqueue_if_busy method."""
+        # Not busy - should not queue
+        was_queued, msg = await manager.enqueue_if_busy("session-1", "Hello")
+        assert not was_queued
+        assert msg is None
+
+        # Now make it busy
+        await manager.acquire("session-1")
+
+        # Should queue now
+        was_queued, msg = await manager.enqueue_if_busy("session-1", "World")
+        assert was_queued
+        assert msg is not None
+        assert msg.prompt == "World"
+
+    @pytest.mark.asyncio
+    async def test_dequeue(self, manager):
+        """Test dequeueing messages."""
+        await manager.enqueue("session-1", "Hello")
+        await manager.enqueue("session-1", "World")
+
+        msg = await manager.dequeue("session-1")
+        assert msg.prompt == "Hello"
+
+        msg = await manager.dequeue("session-1")
+        assert msg.prompt == "World"
+
+        msg = await manager.dequeue("session-1")
+        assert msg is None
+
+    @pytest.mark.asyncio
+    async def test_peek(self, manager):
+        """Test peeking messages."""
+        await manager.enqueue("session-1", "Hello")
+
+        msg = await manager.peek("session-1")
+        assert msg.prompt == "Hello"
+        assert manager.queued_count("session-1") == 1
+
+    @pytest.mark.asyncio
+    async def test_clear_queue(self, manager):
+        """Test clearing a session's queue."""
+        await manager.enqueue("session-1", "1")
+        await manager.enqueue("session-1", "2")
+        await manager.enqueue("session-1", "3")
+
+        count = await manager.clear_queue("session-1")
+        assert count == 3
+        assert manager.queued_count("session-1") == 0
+
+    def test_queued_count_empty(self, manager):
+        """Test queued_count for non-existent session."""
+        assert manager.queued_count("nonexistent") == 0
+
+    def test_queued_prompts_empty(self, manager):
+        """Test queued_prompts for non-existent session."""
+        assert manager.queued_prompts("nonexistent") == []
+
+    @pytest.mark.asyncio
+    async def test_queued_prompts(self, manager):
+        """Test queued_prompts."""
+        await manager.enqueue("session-1", "Hello")
+        await manager.enqueue("session-1", "World")
+
+        prompts = manager.queued_prompts("session-1")
+        assert prompts == ["Hello", "World"]
+
+    @pytest.mark.asyncio
+    async def test_get_queue_status(self, manager):
+        """Test get_queue_status method."""
+        await manager.enqueue("session-1", "Hello")
+        await manager.acquire("session-1")
+
+        status = manager.get_queue_status("session-1")
+        assert status["session_id"] == "session-1"
+        assert status["is_busy"] is True
+        assert status["queued_count"] == 1
+        assert "Hello" in status["queued_prompts"]
+
+    @pytest.mark.asyncio
+    async def test_get_all_status(self, manager):
+        """Test get_all_status method."""
+        await manager.acquire("session-1")
+        await manager.enqueue("session-2", "Hello")
+
+        status = manager.get_all_status()
+        assert "session-1" in status["busy_sessions"]
+        assert status["total_queued"] == 1
+        assert "sessions" in status
+
+
+class TestGlobalQueueManager:
+    """Tests for global queue manager singleton."""
+
+    def test_singleton(self):
+        """Test that get_message_queue_manager returns a singleton."""
+        manager1 = get_message_queue_manager()
+        manager2 = get_message_queue_manager()
+        assert manager1 is manager2
+
+
+class TestRunWithQueue:
+    """Tests for run_with_queue helper function."""
+
+    @pytest.mark.asyncio
+    async def test_processes_immediately_when_not_busy(self):
+        """Test that messages are processed immediately when not busy."""
+        processed = []
+
+        async def processor(prompt, attachments):
+            processed.append(prompt)
+            return f"Result: {prompt}"
+
+        # Use a unique session for this test
+        result = await run_with_queue("run-test-1", "Hello", processor)
+        assert result == "Result: Hello"
+        assert "Hello" in processed
+
+    @pytest.mark.asyncio
+    async def test_returns_none_when_queued(self):
+        """Test that queued messages return None."""
+        manager = get_message_queue_manager()
+
+        # Acquire the session first
+        await manager.acquire("run-test-2")
+
+        async def processor(prompt, attachments):
+            return f"Result: {prompt}"
+
+        try:
+            result = await run_with_queue("run-test-2", "Hello", processor)
+            assert result is None  # Queued, not processed
+        finally:
+            manager.release("run-test-2")
+            # Clean up
+            await manager.clear_queue("run-test-2")

tests/test_multi_agent.py

@@ -0,0 +1,288 @@
+"""Tests for the multi_agent module."""
+
+import pytest
+
+from amcp.multi_agent import (
+    AgentConfig,
+    AgentMode,
+    AgentRegistry,
+    BUILTIN_AGENTS,
+    PRIMARY_SYSTEM_PROMPT,
+    create_subagent_config,
+    get_agent_config,
+    get_agent_registry,
+)
+
+
+class TestAgentMode:
+    """Tests for AgentMode enum."""
+
+    def test_mode_values(self):
+        """Test that mode values are correct."""
+        assert AgentMode.PRIMARY.value == "primary"
+        assert AgentMode.SUBAGENT.value == "subagent"
+
+    def test_mode_comparison(self):
+        """Test mode comparison."""
+        assert AgentMode.PRIMARY != AgentMode.SUBAGENT
+        assert AgentMode.PRIMARY == AgentMode.PRIMARY
+
+
+class TestAgentConfig:
+    """Tests for AgentConfig dataclass."""
+
+    def test_basic_creation(self):
+        """Test basic AgentConfig creation."""
+        config = AgentConfig(
+            name="test_agent",
+            mode=AgentMode.PRIMARY,
+            description="Test agent",
+            system_prompt="You are a test agent.",
+        )
+        assert config.name == "test_agent"
+        assert config.mode == AgentMode.PRIMARY
+        assert config.description == "Test agent"
+        assert config.can_delegate is True  # Default for PRIMARY
+
+    def test_subagent_cannot_delegate(self):
+        """Test that subagents have can_delegate set to False by __post_init__."""
+        config = AgentConfig(
+            name="test_subagent",
+            mode=AgentMode.SUBAGENT,
+            description="Test subagent",
+            system_prompt="You are a test subagent.",
+            can_delegate=True,  # This should be overridden
+        )
+        # __post_init__ should set can_delegate to False for subagents
+        assert config.can_delegate is False
+
+    def test_default_values(self):
+        """Test default values are set correctly."""
+        config = AgentConfig(
+            name="test",
+            mode=AgentMode.PRIMARY,
+            description="Test",
+            system_prompt="Test prompt",
+        )
+        assert config.tools == []
+        assert config.excluded_tools == []
+        assert config.max_steps == 100
+        assert config.parent_agent is None
+
+    def test_get_effective_tools_with_whitelist(self):
+        """Test get_effective_tools with explicit tool whitelist."""
+        config = AgentConfig(
+            name="test",
+            mode=AgentMode.SUBAGENT,
+            description="Test",
+            system_prompt="Test",
+            tools=["read_file", "grep"],
+        )
+        available = ["read_file", "grep", "bash", "write_file"]
+        effective = config.get_effective_tools(available)
+        assert effective == ["read_file", "grep"]
+
+    def test_get_effective_tools_with_exclusions(self):
+        """Test get_effective_tools with tool exclusions."""
+        config = AgentConfig(
+            name="test",
+            mode=AgentMode.PRIMARY,
+            description="Test",
+            system_prompt="Test",
+            tools=[],  # Empty means all
+            excluded_tools=["bash", "write_file"],
+        )
+        available = ["read_file", "grep", "bash", "write_file"]
+        effective = config.get_effective_tools(available)
+        assert "read_file" in effective
+        assert "grep" in effective
+        assert "bash" not in effective
+        assert "write_file" not in effective
+
+    def test_get_effective_tools_whitelist_and_exclusions(self):
+        """Test get_effective_tools with both whitelist and exclusions."""
+        config = AgentConfig(
+            name="test",
+            mode=AgentMode.SUBAGENT,
+            description="Test",
+            system_prompt="Test",
+            tools=["read_file", "grep", "bash"],
+            excluded_tools=["bash"],
+        )
+        available = ["read_file", "grep", "bash", "write_file"]
+        effective = config.get_effective_tools(available)
+        assert effective == ["read_file", "grep"]
+
+
+class TestBuiltinAgents:
+    """Tests for built-in agent configurations."""
+
+    def test_coder_agent(self):
+        """Test coder agent configuration."""
+        assert "coder" in BUILTIN_AGENTS
+        coder = BUILTIN_AGENTS["coder"]
+        assert coder.mode == AgentMode.PRIMARY
+        assert coder.can_delegate is True
+        assert coder.max_steps == 300
+        assert coder.tools == []  # All tools available
+
+    def test_explorer_agent(self):
+        """Test explorer agent configuration."""
+        assert "explorer" in BUILTIN_AGENTS
+        explorer = BUILTIN_AGENTS["explorer"]
+        assert explorer.mode == AgentMode.SUBAGENT
+        assert explorer.can_delegate is False
+        assert "read_file" in explorer.tools
+        assert "grep" in explorer.tools
+        assert "write_file" in explorer.excluded_tools
+        assert "edit_file" in explorer.excluded_tools
+        assert "bash" in explorer.excluded_tools
+
+    def test_planner_agent(self):
+        """Test planner agent configuration."""
+        assert "planner" in BUILTIN_AGENTS
+        planner = BUILTIN_AGENTS["planner"]
+        assert planner.mode == AgentMode.SUBAGENT
+        assert planner.can_delegate is False
+        assert planner.max_steps == 30
+
+    def test_focused_coder_agent(self):
+        """Test focused_coder agent configuration."""
+        assert "focused_coder" in BUILTIN_AGENTS
+        focused = BUILTIN_AGENTS["focused_coder"]
+        assert focused.mode == AgentMode.SUBAGENT
+        assert "write_file" in focused.tools
+        assert "edit_file" in focused.tools
+        assert "bash" in focused.tools
+
+
+class TestAgentRegistry:
+    """Tests for AgentRegistry."""
+
+    def test_init_with_builtin_agents(self):
+        """Test registry is initialized with built-in agents."""
+        registry = AgentRegistry()
+        assert "coder" in registry.list_agents()
+        assert "explorer" in registry.list_agents()
+        assert "planner" in registry.list_agents()
+        assert "focused_coder" in registry.list_agents()
+
+    def test_register_custom_agent(self):
+        """Test registering a custom agent."""
+        registry = AgentRegistry()
+        custom = AgentConfig(
+            name="custom_agent",
+            mode=AgentMode.PRIMARY,
+            description="Custom test agent",
+            system_prompt="Custom prompt",
+        )
+        registry.register(custom)
+        assert "custom_agent" in registry.list_agents()
+        assert registry.get("custom_agent") == custom
+
+    def test_get_nonexistent_agent(self):
+        """Test getting a non-existent agent returns None."""
+        registry = AgentRegistry()
+        assert registry.get("nonexistent") is None
+
+    def test_list_primary_agents(self):
+        """Test listing primary agents."""
+        registry = AgentRegistry()
+        primary = registry.list_primary_agents()
+        assert "coder" in primary
+        assert "explorer" not in primary
+        assert "planner" not in primary
+
+    def test_list_subagents(self):
+        """Test listing subagents."""
+        registry = AgentRegistry()
+        subagents = registry.list_subagents()
+        assert "explorer" in subagents
+        assert "planner" in subagents
+        assert "focused_coder" in subagents
+        assert "coder" not in subagents
+
+    def test_get_subagents_for_delegating_agent(self):
+        """Test getting subagents for an agent that can delegate."""
+        registry = AgentRegistry()
+        subagents = registry.get_subagents_for("coder")
+        assert len(subagents) > 0
+        assert "explorer" in subagents
+
+    def test_get_subagents_for_non_delegating_agent(self):
+        """Test getting subagents for an agent that cannot delegate."""
+        registry = AgentRegistry()
+        subagents = registry.get_subagents_for("explorer")
+        assert subagents == []
+
+
+class TestGlobalFunctions:
+    """Tests for global helper functions."""
+
+    def test_get_agent_registry_singleton(self):
+        """Test that get_agent_registry returns a singleton."""
+        registry1 = get_agent_registry()
+        registry2 = get_agent_registry()
+        assert registry1 is registry2
+
+    def test_get_agent_config(self):
+        """Test get_agent_config helper."""
+        config = get_agent_config("coder")
+        assert config is not None
+        assert config.name == "coder"
+
+    def test_get_agent_config_nonexistent(self):
+        """Test get_agent_config returns None for non-existent agent."""
+        config = get_agent_config("nonexistent")
+        assert config is None
+
+
+class TestCreateSubagentConfig:
+    """Tests for create_subagent_config function."""
+
+    def test_creates_unique_name(self):
+        """Test that subagent config has a unique name."""
+        config1 = create_subagent_config("coder", "Task 1")
+        config2 = create_subagent_config("coder", "Task 2")
+        assert config1.name != config2.name
+        assert config1.name.startswith("task_")
+        assert config2.name.startswith("task_")
+
+    def test_sets_correct_mode(self):
+        """Test that subagent config has SUBAGENT mode."""
+        config = create_subagent_config("coder", "Test task")
+        assert config.mode == AgentMode.SUBAGENT
+
+    def test_sets_parent_agent(self):
+        """Test that parent agent is set correctly."""
+        config = create_subagent_config("coder", "Test task")
+        assert config.parent_agent == "coder"
+
+    def test_cannot_delegate(self):
+        """Test that subagent cannot delegate."""
+        config = create_subagent_config("coder", "Test task")
+        assert config.can_delegate is False
+
+    def test_custom_tools(self):
+        """Test custom tools can be specified."""
+        config = create_subagent_config(
+            "coder",
+            "Test task",
+            tools=["read_file", "grep"],
+        )
+        assert config.tools == ["read_file", "grep"]
+
+    def test_task_description_in_prompt(self):
+        """Test task description is included in system prompt."""
+        config = create_subagent_config("coder", "Analyze the codebase")
+        assert "Analyze the codebase" in config.system_prompt
+
+
+class TestSystemPromptTemplates:
+    """Tests for system prompt templates."""
+
+    def test_primary_system_prompt_has_placeholders(self):
+        """Test that PRIMARY prompt has expected placeholders."""
+        assert "{agent_name}" in PRIMARY_SYSTEM_PROMPT
+        assert "{work_dir}" in PRIMARY_SYSTEM_PROMPT
+        assert "{current_time}" in PRIMARY_SYSTEM_PROMPT