Function make_spawn_agent

src/server/agent_server.py:1288–1329

Build a :data:`SpawnAgent` bound to ``config``. The returned coroutine matches the ``DirectConnectServer.spawn_agent`` contract: ``(session_id, cwd, permission_mode) -> AgentHandle``.

(config: AgentServerConfig | None = None)

def make_spawn_agent(config: AgentServerConfig | None = None):
    """Build a :data:`SpawnAgent` bound to ``config``.

    The returned coroutine matches the ``DirectConnectServer.spawn_agent``
    contract: ``(session_id, cwd, permission_mode) -> AgentHandle``.
    """

    cfg = config or AgentServerConfig()

    async def spawn(session_id: str, cwd: str, perm_mode: str | None) -> AgentHandle:
        loop = asyncio.get_running_loop()
        out_queue: asyncio.Queue[dict | None] = asyncio.Queue()
        sess = _AgentSession(
            session_id=session_id,
            cwd=cwd,
            config=cfg,
            loop=loop,
            out_queue=out_queue,
        )
        # Build the provider/registry/tool_context off the event loop — these
        # touch config/filesystem and must not block the WS pump.
        await loop.run_in_executor(None, lambda: _build_runtime(sess, perm_mode))
        # Wire the permission handler now that tool_context exists.
        if sess.tool_context is not None and sess.init_error is None:
            sess.tool_context.permission_handler = sess.permission_handler
        sess.start()
        sess.emit_init()

        async def messages_from_agent() -> AsyncIterator[dict]:
            while True:
                item = await out_queue.get()
                if item is None:
                    return
                yield item

        return AgentHandle(
            send_to_agent=sess.send_to_agent,
            messages_from_agent=messages_from_agent,
            shutdown=sess.shutdown,
        )

    return spawn


# ─── runtime construction (mirrors entrypoints/headless.py) ───────────────────