Multi-Agent Patterns

Build systems where one agent orchestrates other agents — the "Agents as Tools" pattern on Universal API.

Overview

A multi-agent system lets you break complex tasks into specialized sub-agents, each with their own model, tools, and credentials. An orchestrator agent receives the user's request and delegates to the right specialist.

User → Orchestrator Agent
          ├── call_uapi_agent → Research Agent (web search, reading)
          ├── call_uapi_agent → Image Generator (Stability AI)
          └── call_uapi_agent → Data Analyst (Python, charts)

This follows the Strands SDK "Agents as Tools" pattern, adapted for UAPI's serverless architecture.

The call_uapi_agent Tool

The recommended approach is a native Strands @tool that calls the UAPI streaming agent API directly via HTTPS:

from strands import Agent, tool
import os, json, urllib.request

@tool
def call_uapi_agent(agent_id: str, prompt: str, conversation_id: str = "") -> str:
    """Call another UAPI agent and return its response.

    Args:
        agent_id: The UUID of the UAPI agent to call
        prompt: The message to send to the sub-agent
        conversation_id: Optional ID for multi-turn conversations

    Returns:
        The sub-agent's text response
    """
    bearer_token = os.environ.get("UNIVERSALAPI_BEARER_TOKEN", "")
    url = f"https://stream.api.universalapi.co/agent/{agent_id}/chat"

    payload = {"prompt": prompt}
    if conversation_id:
        payload["conversationId"] = conversation_id

    req = urllib.request.Request(
        url,
        data=json.dumps(payload).encode("utf-8"),
        method="POST",
        headers={
            "Authorization": f"Bearer {bearer_token}",
            "Content-Type": "application/json",
        },
    )

    with urllib.request.urlopen(req, timeout=840) as resp:
        raw = resp.read().decode("utf-8")

    # Parse response — strip metadata markers
    text_lines = []
    for line in raw.strip().split("\n"):
        if line.startswith("__META__") or line.startswith("__METRICS__"):
            continue
        if line.startswith("__COMPLETE__") or line.startswith("__TOOL__"):
            continue
        if line.startswith("__ERROR__"):
            continue
        text_lines.append(line)

    return "\n".join(text_lines).strip() or "Empty response from sub-agent."

Why Native @tool Instead of MCP?

Aspect	MCP chat_with_agent	Native @tool (recommended)
Timeout	2 min (REST API Gateway limit)	15 min (direct to streaming API)
Network hops	4 (Agent → MCP GW → MCP Server → Agent API)	1 (Agent → Streaming API)
Dependencies	Requires universalapi-full MCP server	None — pure Python
Strands-native	No (MCP transport layer)	Yes — standard @tool decorator
Customizable	Limited	Full control — add retries, parsing, etc.

The MCP approach still works for sub-agent calls under 2 minutes, but the native @tool is recommended for reliability and simplicity.

Response Format

The streaming agent API returns text with metadata markers:

Here is the research I found about quantum computing...
[detailed response text]

__META__{"conversationId":"conv-abc123","agentId":"agent-xyz"}__
__METRICS__{"totalCycles":3,"totalTokens":2500,"toolsUsed":["web_search"]}__
__COMPLETE__

The call_uapi_agent tool strips the __META__, __METRICS__, __TOOL__, and __COMPLETE__ markers, returning only the text content.

Parameters

Parameter	Type	Required	Description
agent_id	string	Yes	UUID of the UAPI agent to call
prompt	string	Yes	Message to send to the sub-agent
conversation_id	string	No	Continue a multi-turn conversation

Authentication

The UNIVERSALAPI_BEARER_TOKEN environment variable is automatically injected by the UAPI runtime. It contains the caller's Bearer token, which is passed through to sub-agent calls. No additional setup is needed.

Complete Example

Here's a full orchestrator agent that delegates to two sub-agents:

import os
import json
import urllib.request
from strands import Agent, tool
from strands.models.bedrock import BedrockModel

# ── Sub-Agent Registry ──────────────────────────────────────────
SUB_AGENTS = {
    "researcher": {
        "agent_id": "agent-abc123...",
        "description": "Web research — searches and reads web pages",
    },
    "image-gen": {
        "agent_id": "agent-def456...",
        "description": "Image generation via Stability AI",
    },
}

# ── The Core Tool ───────────────────────────────────────────────
@tool
def call_uapi_agent(agent_id: str, prompt: str, conversation_id: str = "") -> str:
    """Call another UAPI agent and return its response.

    Args:
        agent_id: UUID of the UAPI agent to call
        prompt: Message to send to the sub-agent
        conversation_id: Optional conversation ID for multi-turn

    Returns:
        The sub-agent's text response
    """
    bearer_token = os.environ.get("UNIVERSALAPI_BEARER_TOKEN", "")
    url = f"https://stream.api.universalapi.co/agent/{agent_id}/chat"

    payload = {"prompt": prompt}
    if conversation_id:
        payload["conversationId"] = conversation_id

    try:
        req = urllib.request.Request(
            url,
            data=json.dumps(payload).encode("utf-8"),
            method="POST",
            headers={
                "Authorization": f"Bearer {bearer_token}",
                "Content-Type": "application/json",
            },
        )
        with urllib.request.urlopen(req, timeout=840) as resp:
            raw = resp.read().decode("utf-8")

        text_lines = []
        for line in raw.strip().split("\n"):
            if line.startswith(("__META__", "__METRICS__", "__COMPLETE__",
                                "__TOOL__", "__ERROR__")):
                continue
            text_lines.append(line)

        return "\n".join(text_lines).strip() or "Empty response."
    except Exception as e:
        return f"Error calling sub-agent: {e}"

# ── Orchestrator Agent ──────────────────────────────────────────
def create_agent():
    model = BedrockModel(
        model_id="us.anthropic.claude-sonnet-4-20250514-v1:0",
        region_name=os.environ.get("AWS_REGION", "us-east-1"),
    )

    agent_list = "\n".join(
        f"  - {name} (ID: {info['agent_id']}): {info['description']}"
        for name, info in SUB_AGENTS.items()
    )

    system_prompt = f"""You are a multi-agent orchestrator. Delegate tasks to
specialized sub-agents using call_uapi_agent.

Available sub-agents:
{agent_list}

Guidelines:
- Analyze the request before delegating
- Give sub-agents clear, specific prompts
- Synthesize their responses into a coherent answer
- Answer simple questions directly without delegating
"""

    agent = Agent(
        model=model,
        system_prompt=system_prompt,
        tools=[call_uapi_agent],
        session_manager=session_manager,  # Injected by UAPI runtime
    )
    return agent, []  # No MCP clients needed

Benefits of Multi-Agent on UAPI

Benefit	Description
Independent deployment	Update sub-agents without touching the orchestrator
Separate billing	Each sub-agent call is billed independently
Reusability	The same sub-agent can serve multiple orchestrators
Credential isolation	Sub-agents have their own API keys and MCP tools
Discoverability	Find sub-agents via the UAPI catalog or search API
15-minute timeout	Complex research or generation tasks have plenty of time

Best Practices

1. Write Clear Sub-Agent Prompts

# ❌ Vague
call_uapi_agent(agent_id="...", prompt="AI stuff")

# ✅ Specific
call_uapi_agent(
    agent_id="...",
    prompt="Search for the top 3 AI safety research papers published in 2025. "
           "For each paper, provide the title, authors, and a 2-sentence summary."
)

2. Don't Over-Delegate

If the orchestrator can answer directly, it should. Only delegate when the sub-agent has specialized tools or knowledge the orchestrator lacks.

3. Use Multi-Turn for Follow-Ups

# First call
result = call_uapi_agent(agent_id="...", prompt="Research quantum computing")
# The result includes: [Sub-agent conversationId: conv-abc123]

# Follow-up in the same conversation
result2 = call_uapi_agent(
    agent_id="...",
    prompt="Now focus on error correction techniques",
    conversation_id="conv-abc123"
)

4. Add Error Handling

The call_uapi_agent tool returns error messages as strings (not exceptions), so the orchestrator LLM can reason about failures and retry or try a different approach.

5. Dynamic Discovery

Instead of hardcoding sub-agent IDs, you can use the UAPI search API to find agents dynamically:

@tool
def find_agent(query: str) -> str:
    """Search the UAPI catalog for agents matching a query."""
    url = f"https://api.universalapi.co/search?q={urllib.parse.quote(query)}&type=agent"
    req = urllib.request.Request(url)
    with urllib.request.urlopen(req, timeout=30) as resp:
        data = json.loads(resp.read().decode())
    return json.dumps(data.get("data", {}).get("results", []), indent=2)

Comparison with Strands SDK Patterns

The Strands SDK offers three multi-agent patterns: Graph, Swarm, and Workflow. UAPI's call_uapi_agent maps most closely to the "Agents as Tools" pattern, which is the foundation for all three.

Pattern	Best For	UAPI Equivalent
Agents as Tools	Hierarchical delegation	call_uapi_agent @tool
Graph	Conditional branching, loops	Orchestrator with routing logic
Swarm	Autonomous handoffs between peers	Chain of call_uapi_agent calls
Workflow	Fixed DAG of tasks	Sequential/parallel call_uapi_agent

For more on Strands multi-agent patterns, see the Strands SDK documentation.

Try It

The Multi-Agent Sample agent is live on UAPI:

• Catalog: universalapi.co/agents/snowtimber/multi-agent-sample
• Source code: Visible on the agent's detail page — copy and adapt for your own orchestrator