[eric] agents: native SpawnAgent tool (prompt + run_in_background only) replaces the CLI's built-in Agent; child runs as a real dashboard session

backend/apps/agents/agent_manager.py

@@ -20,6 +20,7 @@ from backend.apps.agents.manager.session.session_store import (
 from backend.apps.agents.manager.streaming.state import ThinkingState, TurnState
 from backend.apps.agents.manager.streaming.PartialReply import PartialReply
 from backend.apps.agents.manager.session.SessionLifecycle import SessionLifecycle
+from backend.apps.agents.manager.SpawnAgentRun import SpawnAgentRun
 from backend.apps.agents.manager.session.SessionPersistence import SessionPersistence
 from backend.apps.agents.manager.Messaging import Messaging
 from backend.apps.agents.manager.SessionControl import SessionControl
@@ -38,7 +39,7 @@ os.environ.setdefault("CLAUDE_CODE_STREAM_CLOSE_TIMEOUT", "3600000")
 MAX_CONCURRENT_TURNS = int(os.environ.get("OSW_MAX_CONCURRENT_TURNS", "8") or "0")
 
 
-class AgentManager(SessionLifecycle, SessionPersistence, Messaging, SessionControl, AgentLaunch, MockAgent, TurnRunner, RunOptions, RunSupport):
+class AgentManager(SessionLifecycle, SessionPersistence, Messaging, SessionControl, AgentLaunch, SpawnAgentRun, MockAgent, TurnRunner, RunOptions, RunSupport):
     @typechecked
     def __init__(self):
         self.sessions: Dict[str, AgentSession] = {}

backend/apps/agents/manager/SpawnAgentRun.py

@@ -0,0 +1,99 @@
+"""spawn_agent: back the SpawnAgent MCP tool with a FRESH sub-agent session (no history
+copy; the prompt must be self-contained). Replaces the CLI's built-in Agent tool, which is
+blocked in RunOptions: its subagent types resolve to models router setups can't serve, and
+its schema drags description/subagent_type/model/isolation along. Mixin, same MRO pattern
+as AgentLaunch."""
+
+import asyncio
+import logging
+from datetime import datetime
+from typing import Dict, Optional
+from uuid import uuid4
+
+from typeguard import typechecked
+
+from backend.apps.agents.core.models import AgentSession, Message
+from backend.apps.agents.core.ws_manager import ws_manager
+from backend.apps.agents.manager.AgentManagerProtocol import AgentManagerProtocol
+from backend.apps.agents.manager.session.apply_context_window import apply_context_window
+from backend.apps.agents.manager.session.session_store import load_session_data
+
+logger = logging.getLogger(__name__)
+
+
+def last_assistant_text(session: AgentSession) -> Optional[str]:
+    for msg in reversed(session.messages):
+        if msg.role == "assistant":
+            content = msg.content
+            if isinstance(content, str):
+                return content
+            if isinstance(content, list):
+                texts = [b.get("text", "") for b in content if isinstance(b, dict) and b.get("type") == "text"]
+                return "\n".join(texts)
+            return str(content)
+    return None
+
+
+class SpawnAgentRun(AgentManagerProtocol):
+    @typechecked
+    async def spawn_agent(
+        self,
+        prompt: str,
+        parent_session_id: str,
+        dashboard_id: Optional[str] = None,
+        run_in_background: bool = False,
+    ) -> Dict:
+        parent = self.sessions.get(parent_session_id)
+        if not parent:
+            data = load_session_data(parent_session_id)
+            if data is None:
+                raise ValueError(f"Parent session {parent_session_id} not found")
+            parent = AgentSession(**data)
+
+        title = (prompt.strip().splitlines() or [""])[0][:60] or "Sub-agent"
+        child = AgentSession(
+            id=uuid4().hex,
+            name=title,
+            status="running",
+            model=parent.model,
+            mode="sub-agent",
+            system_prompt=parent.system_prompt,
+            allowed_tools=list(parent.allowed_tools),
+            max_turns=parent.max_turns or 25,
+            cwd=parent.cwd,
+            created_at=datetime.now(),
+            dashboard_id=dashboard_id or parent.dashboard_id,
+            parent_session_id=parent_session_id,
+        )
+        apply_context_window(child)
+        self.sessions[child.id] = child
+
+        await ws_manager.broadcast_global("agent:status", {
+            "session_id": child.id,
+            "status": child.status,
+            "session": child.model_dump(mode="json"),
+        })
+
+        user_msg = Message(
+            role="user",
+            content=prompt,
+            branch_id=child.active_branch_id,
+        )
+        child.messages.append(user_msg)
+        await ws_manager.send_to_session(child.id, "agent:message", {
+            "session_id": child.id,
+            "message": user_msg.model_dump(mode="json"),
+        })
+
+        if run_in_background:
+            # Fire-and-forget; the child's card carries its progress and result. Keep a handle in self.tasks so stop/shutdown machinery sees it.
+            task = asyncio.create_task(self.run_agent_loop(child.id, prompt))
+            self.tasks[child.id] = task
+            return {"session_id": child.id, "background": True}
+
+        await self.run_agent_loop(child.id, prompt)
+        return {
+            "session_id": child.id,
+            "response": last_assistant_text(child) or "No response from sub-agent.",
+            "cost_usd": child.cost_usd,
+        }

backend/apps/agents/manager/permissions/build_effective_tool_lists.py

@@ -60,6 +60,14 @@ def build_effective_tool_lists(
                         effective_disallowed.append(f"mcp__openswarm-invoke-agent__{it}")
                 continue
 
+            if name == "openswarm-spawn-agent":
+                policy = builtin_perms.get("Agent", "always_allow")
+                if policy == "always_allow":
+                    effective_allowed.append("mcp__openswarm-spawn-agent__SpawnAgent")
+                elif policy == "deny":
+                    effective_disallowed.append("mcp__openswarm-spawn-agent__SpawnAgent")
+                continue
+
             if name == "openswarm-skill":
                 policy = builtin_perms.get("Skill", "always_allow")
                 if policy == "always_allow":

backend/apps/agents/manager/prompt/tool_catalog.py

@@ -19,7 +19,6 @@ FULL_TOOLS = [
     "TaskOutput", "TaskStop",
     "CronCreate", "CronList", "CronDelete",
     "InvokeAgent",
-    "Agent",
     # ToolSearch is the loader the CLI uses to expose deferred tool schemas on demand. Must be in the allowedTools whitelist or the model can't call it, which means none of the deferred extended tools become reachable even when the CLI advertises them in the system prompt.
     "ToolSearch",
 ]

backend/apps/agents/manager/register_builtin_mcp_servers.py

@@ -78,6 +78,23 @@ def register_builtin_mcp_servers(
             "type": "stdio",
         }
 
+    # SpawnAgent replaces the CLI's built-in Agent tool (blocked in RunOptions); gated by the same "Agent" permission so the Tools-page toggle keeps working.
+    if builtin_perms.get("Agent", "always_allow") != "deny":
+        spawn_agent_server_path = os.path.join(
+            agents_dir, "spawn_agent_mcp_server.py"
+        )
+        mcp_servers["openswarm-spawn-agent"] = {
+            "command": sys.executable,
+            "args": [spawn_agent_server_path],
+            "env": {
+                "OPENSWARM_PORT": os.environ.get("OPENSWARM_PORT", "8324"),
+                "OPENSWARM_AUTH_TOKEN": get_auth_token(),
+                "OPENSWARM_PARENT_SESSION_ID": session.id,
+                "OPENSWARM_DASHBOARD_ID": session.dashboard_id or "",
+            },
+            "type": "stdio",
+        }
+
     # Always-on meta-MCP server. Exposes MCPList / MCPSearch / MCPActivate so the model can discover and activate user MCPs at runtime. The activation gate (active_mcps filter in build_mcp_servers above) ensures the model cannot reach any other MCP server's tools without going through this layer first.
     mcp_meta_server_path = os.path.join(
         agents_dir, "mcp_meta_server.py"

backend/apps/agents/manager/run/RunOptions.py

@@ -229,6 +229,8 @@ class RunOptions(AgentManagerProtocol):
         # The claude_code preset auto-attaches the user's claude.ai- connected partner MCPs (`mcp__claude_ai_*`). Those bypass our MCPActivate gate, don't share OAuth state with the OpenSwarm Gmail/Calendar/Drive connectors the user actually configured here, and confuse the model into picking the partner shim instead of our vetted server. Hard-block them at the SDK layer so the model can't even attempt the call.
         options_kwargs["disallowed_tools"] = [
             "mcp__claude_ai_*",
+            # The CLI's built-in Agent tool is replaced by our SpawnAgent MCP (prompt + run_in_background only); its subagent types resolve to models router setups can't serve.
+            "Agent",
         ]
 
         if session.cwd:

backend/apps/agents/spawn_agent_mcp_server.py

@@ -0,0 +1,163 @@
+#!/usr/bin/env python3
+"""Stdio MCP server exposing the SpawnAgent tool; proxies to /api/spawn-agent/run.
+
+Replaces the CLI's built-in Agent tool (blocked in RunOptions): that schema drags
+description/subagent_type/model/isolation along, and its subagent types resolve to
+models our router setups can't serve. This one takes prompt + run_in_background,
+nothing else; the child runs as a real OpenSwarm session card on the dashboard."""
+
+import json
+import sys
+import os
+import urllib.request
+import urllib.error
+
+BACKEND_PORT = os.environ.get("OPENSWARM_PORT", "8324")
+BACKEND_AUTH = os.environ.get("OPENSWARM_AUTH_TOKEN", "")
+BACKEND_URL = f"http://127.0.0.1:{BACKEND_PORT}/api/spawn-agent/run"
+PARENT_SESSION_ID = os.environ.get("OPENSWARM_PARENT_SESSION_ID", "")
+DASHBOARD_ID = os.environ.get("OPENSWARM_DASHBOARD_ID", "")
+
+TOOLS = [
+    {
+        "name": "SpawnAgent",
+        "description": (
+            "Spawn a sub-agent to handle a task. The sub-agent runs as its own "
+            "agent session (visible on the dashboard) with the same working "
+            "directory and model as you. By default this blocks until the "
+            "sub-agent finishes and returns its final answer; set "
+            "run_in_background=true to return immediately and let it work on "
+            "its own, its progress and result appear on its dashboard card."
+        ),
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "prompt": {
+                    "type": "string",
+                    "description": (
+                        "The task for the sub-agent. Include all context it "
+                        "needs; it does not see your conversation."
+                    ),
+                },
+                "run_in_background": {
+                    "type": "boolean",
+                    "description": (
+                        "true = return immediately with the sub-agent's session "
+                        "id instead of waiting for its result."
+                    ),
+                },
+            },
+            "required": ["prompt"],
+        },
+    },
+]
+
+
+def send_response(id_, result=None, error=None):
+    msg = {"jsonrpc": "2.0", "id": id_}
+    if error is not None:
+        msg["error"] = error
+    else:
+        msg["result"] = result
+    sys.stdout.write(json.dumps(msg) + "\n")
+    sys.stdout.flush()
+
+
+def call_backend(prompt: str, run_in_background: bool) -> dict:
+    payload = json.dumps({
+        "prompt": prompt,
+        "run_in_background": run_in_background,
+        "parent_session_id": PARENT_SESSION_ID,
+        "dashboard_id": DASHBOARD_ID,
+    }).encode()
+    headers = {"Content-Type": "application/json"}
+    if BACKEND_AUTH:
+        headers["Authorization"] = f"Bearer {BACKEND_AUTH}"
+    req = urllib.request.Request(
+        BACKEND_URL,
+        data=payload,
+        headers=headers,
+        method="POST",
+    )
+    try:
+        with urllib.request.urlopen(req, timeout=1800) as resp:
+            return json.loads(resp.read().decode())
+    except urllib.error.HTTPError as e:
+        body = e.read().decode() if e.fp else str(e)
+        return {"error": f"HTTP {e.code}: {body}"}
+    except Exception as e:
+        return {"error": str(e)}
+
+
+def handle_tool_call(tool_name: str, arguments: dict) -> dict:
+    if tool_name != "SpawnAgent":
+        return {"content": [{"type": "text", "text": f"Unknown tool: {tool_name}"}], "isError": True}
+
+    prompt = arguments.get("prompt", "")
+    run_in_background = bool(arguments.get("run_in_background", False))
+
+    if not prompt:
+        return {"content": [{"type": "text", "text": "Error: prompt is required"}], "isError": True}
+
+    result = call_backend(prompt, run_in_background)
+
+    if "error" in result:
+        return {"content": [{"type": "text", "text": f"Error: {result['error']}"}], "isError": True}
+
+    sid = result.get("session_id", "")
+    if run_in_background:
+        return {"content": [{"type": "text", "text": (
+            f"Spawned background sub-agent (session: {sid}). It is working on its own "
+            "dashboard card; its result will appear there. Do not wait for it."
+        )}]}
+
+    response = result.get("response", "No response from sub-agent.")
+    cost = result.get("cost_usd", 0)
+    lines = [f"**Sub-Agent Result** (session: {sid})"]
+    if cost > 0:
+        lines.append(f"*Cost: ${cost:.4f}*")
+    lines.append("")
+    lines.append(response)
+    return {"content": [{"type": "text", "text": "\n".join(lines)}]}
+
+
+def main():
+    for line in sys.stdin:
+        line = line.strip()
+        if not line:
+            continue
+        try:
+            msg = json.loads(line)
+        except json.JSONDecodeError:
+            continue
+
+        method = msg.get("method")
+        id_ = msg.get("id")
+        params = msg.get("params", {})
+
+        if method == "initialize":
+            send_response(id_, {
+                "protocolVersion": "2024-11-05",
+                "capabilities": {"tools": {}},
+                "serverInfo": {
+                    "name": "openswarm-spawn-agent",
+                    "version": "1.0.0",
+                },
+            })
+        elif method == "notifications/initialized":
+            pass
+        elif method == "tools/list":
+            send_response(id_, {"tools": TOOLS})
+        elif method == "tools/call":
+            tool_name = params.get("name", "")
+            arguments = params.get("arguments", {})
+            result = handle_tool_call(tool_name, arguments)
+            send_response(id_, result)
+        elif method == "ping":
+            send_response(id_, {})
+        elif id_ is not None:
+            send_response(id_, error={"code": -32601, "message": f"Method not found: {method}"})
+
+
+if __name__ == "__main__":
+    main()

backend/main.py

@@ -870,6 +870,37 @@ async def settings_meta(action: str, request: Request):
 
 
 
+@app.post("/api/spawn-agent/run")
+async def spawn_agent_run(request: Request):
+    """Spawn a fresh sub-agent session for the SpawnAgent tool.
+    Called by the spawn_agent_mcp_server stdio subprocess."""
+    body = await request.json()
+    prompt = body.get("prompt", "")
+    parent_session_id = body.get("parent_session_id", "")
+    dashboard_id = body.get("dashboard_id", "")
+    run_in_background = bool(body.get("run_in_background", False))
+
+    if not prompt:
+        return JSONResponse({"error": "prompt is required"}, status_code=400)
+    if not parent_session_id:
+        return JSONResponse({"error": "parent_session_id is required"}, status_code=400)
+
+    try:
+        from backend.apps.agents.agent_manager import agent_manager
+        result = await agent_manager.spawn_agent(
+            prompt=prompt,
+            parent_session_id=parent_session_id,
+            dashboard_id=dashboard_id or None,
+            run_in_background=run_in_background,
+        )
+        return JSONResponse(result)
+    except ValueError as e:
+        return JSONResponse({"error": str(e)}, status_code=404)
+    except Exception as e:
+        logger.exception("spawn_agent_run failed")
+        return JSONResponse({"error": str(e)}, status_code=500)
+
+
 @app.post("/api/invoke-agent/run")
 async def invoke_agent_run(request: Request):
     """Fork an existing agent session and send it a new message.

backend/tests/test_spawn_agent.py

@@ -0,0 +1,107 @@
+"""SpawnAgent: the native replacement for the CLI's built-in Agent tool. The child is a
+fresh session inheriting the parent's model/cwd/dashboard; sync waits and returns the last
+assistant text, background returns immediately. The builtin stays blocked so the model only
+ever sees the two-arg schema."""
+import asyncio
+import json
+import subprocess
+import sys
+from typing import Dict, List, Optional
+
+from pytest import MonkeyPatch, raises
+
+from backend.apps.agents.agent_manager import agent_manager
+from backend.apps.agents.core.models import AgentSession, Message
+
+
+def seed_parent() -> AgentSession:
+    parent = AgentSession(name="parent", model="opus-4-8", cwd="/tmp/pw", dashboard_id="dashX")
+    agent_manager.sessions[parent.id] = parent
+    return parent
+
+
+def test_spawn_agent_sync_returns_child_answer(monkeypatch: MonkeyPatch) -> None:
+    parent = seed_parent()
+
+    async def fake_loop(session_id: str, prompt: str, **kwargs: object) -> None:
+        s = agent_manager.sessions[session_id]
+        s.messages.append(Message(role="assistant", content="child says done", branch_id=s.active_branch_id))
+        s.status = "completed"
+
+    monkeypatch.setattr(agent_manager, "run_agent_loop", fake_loop)
+    result = asyncio.run(agent_manager.spawn_agent(prompt="do the thing", parent_session_id=parent.id))
+
+    child = agent_manager.sessions[result["session_id"]]
+    assert result["response"] == "child says done"
+    assert child.mode == "sub-agent"
+    assert child.parent_session_id == parent.id
+    assert child.model == parent.model
+    assert child.cwd == parent.cwd
+    assert child.dashboard_id == "dashX"
+    assert child.messages[0].role == "user" and child.messages[0].content == "do the thing"
+
+
+def test_spawn_agent_background_returns_immediately(monkeypatch: MonkeyPatch) -> None:
+    parent = seed_parent()
+    started: List[str] = []
+
+    async def slow_loop(session_id: str, prompt: str, **kwargs: object) -> None:
+        started.append(session_id)
+        await asyncio.sleep(30)
+
+    monkeypatch.setattr(agent_manager, "run_agent_loop", slow_loop)
+
+    async def run() -> Dict:
+        result = await asyncio.wait_for(
+            agent_manager.spawn_agent(prompt="long task", parent_session_id=parent.id, run_in_background=True),
+            timeout=2.0,
+        )
+        await asyncio.sleep(0.05)
+        agent_manager.tasks[result["session_id"]].cancel()
+        return result
+
+    result = asyncio.run(run())
+    assert result["background"] is True
+    assert started == [result["session_id"]]
+
+
+def test_spawn_agent_unknown_parent_raises() -> None:
+    with raises(ValueError):
+        asyncio.run(agent_manager.spawn_agent(prompt="x", parent_session_id="nope-" + "0" * 28))
+
+
+def test_builtin_agent_tool_stays_blocked() -> None:
+    # The CLI's Agent tool must not be offered: out of the catalog AND hard-blocked at the SDK layer.
+    from backend.apps.agents.manager.prompt.tool_catalog import FULL_TOOLS
+    assert "Agent" not in FULL_TOOLS
+    import inspect
+    from backend.apps.agents.manager.run import RunOptions
+    src = inspect.getsource(RunOptions)
+    assert '"Agent",' in src.split('disallowed_tools"] = [')[1][:300]
+
+
+def test_spawn_server_schema_is_prompt_plus_background_only() -> None:
+    from backend.apps.agents import spawn_agent_mcp_server as srv
+    tool = srv.TOOLS[0]
+    assert tool["name"] == "SpawnAgent"
+    assert set(tool["inputSchema"]["properties"].keys()) == {"prompt", "run_in_background"}
+    assert tool["inputSchema"]["required"] == ["prompt"]
+
+
+def test_spawn_server_speaks_mcp_stdio() -> None:
+    proc = subprocess.Popen(
+        [sys.executable, "backend/apps/agents/spawn_agent_mcp_server.py"],
+        stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True,
+    )
+    try:
+        msgs = [
+            {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}},
+            {"jsonrpc": "2.0", "method": "notifications/initialized"},
+            {"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}},
+        ]
+        out, _ = proc.communicate("\n".join(json.dumps(m) for m in msgs) + "\n", timeout=15)
+        lines = [json.loads(line) for line in out.strip().splitlines()]
+        assert lines[0]["result"]["serverInfo"]["name"] == "openswarm-spawn-agent"
+        assert lines[1]["result"]["tools"][0]["name"] == "SpawnAgent"
+    finally:
+        proc.kill()

frontend/src/app/pages/AgentChat/parsing/agentToolParsing.ts

@@ -13,7 +13,9 @@ export function isInvokeAgentTool(name: string): boolean {
 }
 
 export function isCreateAgentTool(name: string): boolean {
-  return name === 'Agent';
+  if (name === 'Agent') return true;
+  const mcp = parseMcpToolName(name);
+  return mcp.isMcp && mcp.serverSlug === 'openswarm-spawn-agent';
 }
 
 export function parseInvokedSessionId(rawText: string): string | null {