CugaSupervisor

The CugaSupervisor class coordinates multiple agents: it receives a user task, delegates work to specialized sub-agents, and returns a final answer. You can mix local CugaAgent instances with remote agents via the A2A protocol.

Try the Demo

The bundled CRM + email multi-agent demo can be launched with:

cuga start demo_supervisor

This brings up the same demo surface as demo_crm but with the supervisor wired to a CRM sub-agent and an email sub-agent. Use it to see delegation and variable-passing end to end before building your own configuration.

Quick Start

from cuga import CugaAgent, CugaSupervisor
from langchain_core.tools import tool
import asyncio

@tool
def get_customers(limit: int = 10) -> str:
    """Fetch top customers from CRM with name, email, and revenue. Returns a formatted string."""
    customers = [
        "Alice ([email protected], $250,000)",
        "Bob ([email protected], $180,000)",
        "Carol ([email protected], $120,000)",
        "Dave ([email protected], $95,000)",
        "Eve ([email protected], $88,000)",
    ]
    top = customers[: min(limit, len(customers))]
    return "Top customers by revenue: " + "; ".join(f"{i+1}. {c}" for i, c in enumerate(top))

@tool
def send_email(to: str, body: str) -> str:
    """Send an email. Returns confirmation."""
    return f"Email sent successfully to {to}"

crm_agent = CugaAgent(tools=[get_customers])
crm_agent.description = "CRM and customer data"

email_agent = CugaAgent(tools=[send_email])
email_agent.description = "Sending emails and notifications"

supervisor = CugaSupervisor(agents={
    "crm": crm_agent,
    "email": email_agent,
})

result = await supervisor.invoke(
    "Get our top 5 customers by revenue, then send the top customer a thank-you email"
)
print(result.answer)

Initialization

Core Methods

invoke

Run the supervisor with a user message and get back a final answer.

result = await supervisor.invoke("Get top 5 customers and email the top one a thank-you")
print(result.answer)
print(result.thread_id)

from_yaml

Load supervisor and agents from a YAML configuration file. The supervisor's own model is taken from system settings. Each local sub-agent can configure its own model in YAML.

supervisor = await CugaSupervisor.from_yaml("supervisor_config.yaml")
result = await supervisor.invoke("Your task here")

Sub-agent models in YAML

supervisor:
  name: sales-supervisor
  special_instructions: "Delegate to the appropriate specialist."

agents:
  - name: crm
    description: "Customer records"
    apps: [crm]
    model:
      provider: openai
      model_name: gpt-4o
      max_tokens: 8192

provider selects the sub-agent's backend; model_name selects its model when no higher-priority override applies. For OpenAI and other providers supporting the global override, MODEL_NAME from .env or the process environment wins over YAML model_name. Remove that global override when different sub-agents need different model names. Export the matching provider credentials. CugaSupervisor.from_yaml() does not set the supervisor model from this block. To supply the supervisor's own model directly, construct CugaSupervisor(agents=..., model=...).

Registry and composition in Manage

Create named agents, publish their individual tools/models/policies, then compose a supervisor from those agents in Manage. Referenced sub-agents use their own published configuration. Keep policies on the agent that should enforce them; playbooks and other policy collections are isolated per agent.

Caller identity and child checkpoints

In a multi-user SDK service, pass a trusted caller identity:

result = await supervisor.invoke(
    "Look up my customer records",
    thread_id="conversation-123",
    user_id="authenticated-user-id",
)

Child checkpoints are scoped by caller, parent thread, supervisor, and sub-agent memory scope. Obtain user_id from your authentication layer. Server integrations derive caller identity from the authenticated request.

For a local ACP subprocess rather than a remote A2A endpoint, see CUGA over ACP.

External Agents (A2A)

To expose this CUGA so others call you over A2A, see CUGA as A2A.

To call a remote agent that already speaks A2A, pass an external config in agents:

remote_config = {
    "name": "analytics",
    "type": "external",
    "description": "Remote analytics service",
    "config": {
        "a2a_protocol": {
            "endpoint": "http://localhost:9999",
            "transport": "http",
        },
    },
}

supervisor = CugaSupervisor(agents={
    "crm": crm_agent,
    "email": email_agent,
    "analytics": remote_config,
})

Set pass_variables_a2a = true under [supervisor] in settings.toml if you want to pass variables to A2A agents.

Configuration

When using the CUGA server, enable the supervisor in settings.toml:

[supervisor]
enabled = true
config_path = "path/to/supervisor_config.yaml"  # optional; used when loading from YAML via server
pass_variables_a2a = false  # set true to pass variables to A2A agents in metadata

For SDK-only usage, building the supervisor in code (as in Quick Start) does not require any settings.toml changes.

Policies

CugaSupervisor uses the same policy API as CugaAgent via supervisor.policies. Intent guards block unsafe requests before delegation; playbooks inject multi-agent workflows into the supervisor prompt.

await supervisor.policies.add_intent_guard(
    name="Block Deletes",
    keywords=["delete", "remove"],
    response="Deletion is not permitted.",
)

await supervisor.policies.add_playbook(
    name="Onboarding",
    keywords=["onboard"],
    content="# Onboarding\n1. Delegate to crm to fetch the customer\n2. Delegate to email to send welcome mail",
)

result = await supervisor.invoke("Onboard the new customer Acme Corp")