When are we expecting to see subagents supported in the agent-server?

cc @modern-fireman-99755 may have more info

Sorry for the late answer. Actually i don’t know what there is precisely in the agent server. I suppose you could already use subagents via the delegation tool. But i will double check.

I don't think delegation is enabled there

Perhaps I didn't get the question but I think that you can use subagents in the agent-server. 1. Client side (via RemoteConversation) sends a StartConversationRequest with: - tool_module_qualnames: {"DelegateTool": "openhands.tools.delegate"} — tells the server to import the tool - agent_definitions: [...] — registers sub-agent definitions so DelegateTool can see them 2. Server side dynamically imports the tool module and registers agent definitions. (see openhands-agent-server/models.py:123) Subagents, for instance, are not present in the Cloud, but they are available in the CLI.

ooh interesting. Let me see if I can get this working

I have a working example. I think you will need to call

register_builtins_agents()

before calling the agent-server. The method will expose the builtin agents for the agent-server. The example is working for

software-agent-sdk

branch:

main

Do you have a specific use case in mind for using subagents with the agent-server? Asking because I can see a use case where you spin up the agent-server and then use another language to ping it (or ping it directly). We would likely need a way to tell the agent-server to register the built-in subagents when starting a new conversation, or?

"""
Delegation with Agent Server Example

Minimal example demonstrating DelegateTool used via RemoteConversation
(agent server). The main agent delegates two tasks to sub-agents that
run in parallel on the server side.
"""

import os
import subprocess
import sys
import tempfile
import threading
import time

from pydantic import SecretStr

from openhands.sdk import (
    LLM,
    Agent,
    AgentContext,
    Conversation,
    RemoteConversation,
    Tool,
    get_logger,
)
from openhands.sdk.context import Skill
from openhands.sdk.subagent import register_agent
from openhands.sdk.tool import register_tool
from openhands.tools import register_builtins_agents
from openhands.tools.delegate import DelegateTool


logger = get_logger(__name__)

def _stream_output(stream, prefix, target_stream):
    try:
        for line in iter(stream.readline, ""):
            if line:
                target_stream.write(f"[{prefix}] {line}")
                target_stream.flush()
    except Exception:
        pass
    finally:
        stream.close()


class ManagedAPIServer:
    def __init__(self, port: int = 8000, host: str = "127.0.0.1"):
        self.port = port
        self.host = host
        self.process = None
        self.base_url = f"http://{host}:{port}"

    def __enter__(self):
        print(f"Starting agent server on {self.base_url}...")
        self.process = subprocess.Popen(
            [
                "python", "-m", "openhands.agent_server",
                "--port", str(self.port),
                "--host", self.host,
            ],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            env={"LOG_JSON": "true", **os.environ},
        )
        threading.Thread(
            target=_stream_output,
            args=(self.process.stdout, "SERVER", sys.stdout),
            daemon=True,
        ).start()
        threading.Thread(
            target=_stream_output,
            args=(self.process.stderr, "SERVER", sys.stderr),
            daemon=True,
        ).start()

        import httpx

        for _ in range(30):
            try:
                if httpx.get(f"{self.base_url}/health", timeout=1.0).status_code == 200:
                    print(f"Agent server ready at {self.base_url}")
                    return self
            except Exception:
                pass
            if self.process.poll() is not None:
                raise RuntimeError("Server process terminated unexpectedly.")
            time.sleep(1)
        raise RuntimeError("Server failed to start after 30 seconds")

    def __exit__(self, *args):
        if self.process:
            self.process.terminate()
            try:
                self.process.wait(timeout=5)
            except subprocess.TimeoutExpired:
                self.process.kill()
                self.process.wait()
            time.sleep(0.5)
            print("Agent server stopped.")


api_key = os.getenv("LLM_API_KEY")
assert api_key, "LLM_API_KEY environment variable is not set."

llm = LLM(
    model=os.getenv("LLM_MODEL", "openhands/claude-sonnet-4-5-20250929"),
    api_key=SecretStr(api_key),
    base_url=os.getenv("LLM_BASE_URL"),
    usage_id="agent",
)


def create_lodging_planner(llm: LLM) -> Agent:
    return Agent(
        llm=llm,
        tools=[],
        agent_context=AgentContext(
            skills=[
                Skill(
                    name="lodging_planning",
                    content=(
                        "You specialize in finding great places to stay in London. "
                        "Provide 3 hotel recommendations with neighborhoods and "
                        "quick pros/cons."
                    ),
                    trigger=None,
                )
            ],
            system_message_suffix="Focus only on London lodging recommendations.",
        ),
    )


def create_activities_planner(llm: LLM) -> Agent:
    return Agent(
        llm=llm,
        tools=[],
        agent_context=AgentContext(
            skills=[
                Skill(
                    name="activities_planning",
                    content=(
                        "You design concise London itineraries. Suggest 2-3 daily "
                        "highlights grouped by proximity."
                    ),
                    trigger=None,
                )
            ],
            system_message_suffix="Plan practical days in London.",
        ),
    )


# Register sub-agents and tool (client-side; forwarded to server automatically)
register_agent(
    name="lodging_planner",
    factory_func=create_lodging_planner,
    description="Finds London lodging options.",
)
register_agent(
    name="activities_planner",
    factory_func=create_activities_planner,
    description="Creates London activity itineraries.",
)
register_builtins_agents()

# NOTE: DelegateTool auto-registers as "delegate" (its canonical name) when
# its module is imported. We must use that same name in Tool(name=...).
# Registering under a different name (e.g. "DelegateTool") would work locally
# but fail on the agent server, because the server re-imports the module which
# only registers the canonical name.
register_tool(DelegateTool.name, DelegateTool)


with ManagedAPIServer(port=8002) as server:
    from openhands.sdk import Workspace

    temp_dir = tempfile.mkdtemp(prefix="delegation_demo_")
    workspace = Workspace(host=server.base_url, working_dir=temp_dir)

    main_agent = Agent(
        llm=llm,
        tools=[Tool(name=DelegateTool.name)],
    )

    conversation = Conversation(
        agent=main_agent,
        workspace=workspace,
    )
    assert isinstance(conversation, RemoteConversation), (
        "Expected RemoteConversation when workspace has a host"
    )

    print("=" * 70)
    print("Sending delegation request to agent server...")
    print("=" * 70)

    conversation.send_message(
        "Plan a short trip to London. Use the delegation tool to handle "
        "lodging and activities in parallel via the two registered sub-agents. "
        "Keep responses concise (a few bullet points each). "
        "Once you have results from both, give me a short combined summary."
    )
    conversation.run()

    cost = conversation.conversation_stats.get_combined_metrics().accumulated_cost
    print(f"\nEXAMPLE_COST: {cost}")
    print("Done!")

    conversation.close()

My use case is the typical one: I want to ask my agent to do "X, Y, and Z", and if they're separable tasks, the agent should create subagents to do each one in parallel

Make sense. I will take care of that.