usf-agents 1.0.0.post2

A lightweight multi-agent orchestration framework with better control, easy to use for complex to simple use cases. Developer friendly with more visibility and supports all models with OpenAI compatible API.

USF Agents SDK (Python)

Production-grade, OpenAI-compatible Agents SDK for planning, tool execution, and multi‑agent orchestration on the official USF Agent APIs.

This release focuses on a simpler, more powerful developer experience:

• Define tool schemas from docstrings (Google/NumPy) or an optional YAML block — no verbose JSON required.
• Use the lightweight @tool decorator for defaults (name/alias/description) and optionally provide the full schema there too.
• Register tools in one line, batch-register multiple functions, or discover from a module.
• Compose sub-agents with a single call and auto-execute to a final answer with robust policies.

Contents

• Features
• Configuration Reference
• Installation
• Requirements
• Quickstart (copy/paste runnable)
• Tool Definitions Made Simple (Docstrings, YAML, @tool, strictness, explicit schema)
• Auto Execution Modes
• Multi-Agent in Fewer Lines
• Registry Option (alternate flow)
• Advanced Cookbook (end-to-end)
• Migration Notes (breaking changes)
• Troubleshooting / FAQ
• Contributing / License

---

Features

• Docstring-driven tool schemas
  • Precedence: YAML block (OpenAPI-like) → Google-style Args → NumPy-style Parameters
  • Required-parameter equality with signature automatically enforced
• @tool decorator
  • Provide defaults (name, alias, description)
  • Optionally provide the full OpenAI tool schema on the decorator
• Batch registration of tools
  • add_function_tools([...])
  • add_function_tools_from_module(module, filter=...)
• One-call auto execution (plan → tool_calls → final) with policy modes
  • disable | auto | agent | tool
• Multi-agent (manager + sub-agents) in one or two lines each
• Per-agent tool name uniqueness validation
• Strong sequencing safety utilities (unchanged)
• Tracing, workflows, and visualization (unchanged)

---

Configuration Reference

Global config for USFAgent (defaults in parentheses)

• api_key: str (required)
• model: str = 'usf-mini'
• provider: Optional[str] = None (planning/tool-calling only; one of: openrouter, openai, claude, huggingface-inference, groq)
• introduction: str = ''
• knowledge_cutoff: str = '15 January 2025'
• stream: bool = False
• max_loops: int = 20 (1..100)
• backstory: str = ''
• goal: str = ''
• temp_memory:
  • enabled: bool = False
  • max_length: int = 10
  • auto_trim: bool = True
• debug: bool = False

Example (global + stage overrides)

# run: USF_API_KEY=YOUR_KEY python - <<'PY'
import os, asyncio
from usf_agents import USFAgent

async def main():
    agent = USFAgent({
        "api_key": os.getenv("USF_API_KEY"),
        "model": "usf-mini",
        "backstory": "Power user of the system.",
        "goal": "Concise, accurate answers.",
        "temp_memory": {"enabled": True, "max_length": 5, "auto_trim": True},
        "max_loops": 10,
        "planning": {
            "provider": "openai",
            "model": "gpt-4o-mini",
            "introduction": "You are planning the steps to solve the task.",
            "knowledge_cutoff": "15 January 2025",
            "debug": True
        },
        "tool_calling": {
            "provider": "openai",
            "model": "gpt-4o-mini",
            "debug": True
        },
        "final_response": {
            "model": "usf-mini",
            "temperature": 0.5,
            # Any OpenAI Chat Completions params not reserved by USF are passed through
            "max_tokens": 512,
            "top_p": 1.0
        }
    })

    async for chunk in agent.run("Say hello world"):
        if chunk["type"] == "final_answer":
            print("Final:", chunk["content"]); break

asyncio.run(main())
# PY

Stage configs (optional, override fallbacks)

• planning: Dict[str, Any]
  • Supports: api_key, provider, model, introduction, knowledge_cutoff, temperature, stop, debug, ...
• tool_calling: Dict[str, Any]
  • Supports: api_key, provider, model, introduction, knowledge_cutoff, temperature, stop, debug, ...
• final_response: Dict[str, Any]
  • Supports: api_key, model, temperature, stop, date_time_override, debug, ...
  • Also passes through OpenAI Chat Completions params via extra kwargs: response_format, max_tokens, top_p, presence_penalty, frequency_penalty, logit_bias, seed, user, stream_options
  • Note: Do not include "tools" in final_response; tools are not used in the final stage.

Provider usage (planning/tool-calling)

• Default engine is "usf-mini" when no provider is specified.
• For custom/non-default engines, set provider to one of: openrouter, openai, claude, huggingface-inference, groq.
• Omit provider (or leave blank) for default or directly supported models.

Example (provider on planning/tool_calling)

# run: USF_API_KEY=YOUR_KEY python - <<'PY'
import os, asyncio
from usf_agents import USFAgent

async def main():
    agent = USFAgent({
        "api_key": os.getenv("USF_API_KEY"),
        "planning": {"provider": "openai", "model": "gpt-4o-mini"},
        "tool_calling": {"provider": "openai", "model": "gpt-4o-mini"},
        # final_response uses OpenAI-compatible SDK; provider is not required there
    })
    async for chunk in agent.run("Plan then answer without tools"):
        if chunk["type"] == "final_answer":
            print("Final:", chunk["content"]); break

asyncio.run(main())
# PY

Per-run options (passed to agent.run(messages, options))

• tools: List[Tool] (OpenAI function tool format)
• planning/tool_calling/final_response: per-call overrides
• temperature, stop, date_time_override, debug, max_loops

Example (per-run overrides; no tools to ensure final output)

# run: USF_API_KEY=YOUR_KEY python - <<'PY'
import os, asyncio
from usf_agents import USFAgent

async def main():
    agent = USFAgent({"api_key": os.getenv("USF_API_KEY")})
    opts = {
      "temperature": 0.2,
      "max_loops": 5,
      "final_response": {
        "date_time_override": {
          "enabled": True,
          "date": "08/31/2025",
          "time": "07:00:00 AM",
          "timezone": "UTC"
        }
      }
    }
    async for chunk in agent.run("Reply with the current date/time string from system context.", opts):
        if chunk["type"] == "final_answer":
            print("Final:", chunk["content"]); break

asyncio.run(main())
# PY

Date/time override (final response stage)

• final_response.date_time_override: { enabled: bool, date: 'MM/DD/YYYY', time: 'HH:MM:SS AM/PM', timezone: 'IANA or label' }

Example (snippet)

opts = {
  "final_response": {
    "date_time_override": {
      "enabled": True,
      "date": "12/31/2025",
      "time": "11:59:59 PM",
      "timezone": "America/Los_Angeles"
    }
  }
}

Tool schema (OpenAI function-call compatible)

• Tool = {'type': 'function','function': {'name': str,'description': str,'parameters': JSONSchema}}

Example

tool = {
  "type": "function",
  "function": {
    "name": "http_get",
    "description": "Fetch a URL",
    "parameters": {
      "type": "object",
      "properties": {"url": {"type": "string", "description": "URL to fetch"}},
      "required": ["url"]
    }
  }
}

Tool call format (from assistant)

• {'id': '...', 'type': 'function', 'function': {'name': '...', 'arguments': JSON-string}}

Tool result message (from developer)

• {'role': 'tool','tool_call_id': '...', 'name': 'tool_name','content': JSON-string}

Example (manual loop, conceptual)

assistant_tool_calls = {
  "role": "assistant",
  "tool_calls": [
    {"id": "call_1", "type": "function", "function": {"name": "http_get", "arguments": "{\"url\":\"https://example.com\"}"}}
  ],
  "type": "tool_calls"
}
tool_result = {
  "role": "tool",
  "tool_call_id": "call_1",
  "name": "http_get",
  "content": "{\"status\":200,\"body\":\"ok\"}"
}
# Append both to messages before calling agent.run() again.

Multi-agent AgentSpec (ManagerAgent/SubAgent)

• id: str
• name: str
• agent_type: 'manager'|'sub'|'generic'
• backstory: str (optional)
• goal: str (optional)
• context_mode: 'NONE'|'AGENT_DECIDED'|'ALWAYS_FULL'|'CONTEXT_PARAM'
• usf_config: USFAgent config (same shape as constructor)
• tools: List[Tool] (native tools for this wrapper, optional)

Manager/SubAgent APIs

• add_sub_agent(sub: BaseAgentWrapper, schema: Dict, alias: Optional[str]) -> None
• list_tools() -> List[Tool] (native + sub-agents as tools)
• delegate(sub_id, task, policy='inherit_manager_policy', context_param=None, calling_context=None, options=None)
• SubAgent.execute_as_tool(tool_call, calling_context, context_param=None, options=None)
• SubAgent.execute_as_tool_until_final(tool_call, calling_context, context_param=None, options=None)

Example (compose and delegate)

# run: USF_API_KEY=YOUR_KEY python - <<'PY'
import os, asyncio
from usf_agents.multi_agent.base import ManagerAgent, SubAgent

async def main():
    api_key = os.getenv("USF_API_KEY")
    mgr = ManagerAgent({"id":"mgr","name":"Manager","agent_type":"manager","usf_config":{"api_key":api_key}})

    calc = SubAgent({
      "id":"calc","name":"Calculator","agent_type":"sub","context_mode":"NONE",
      "description":"Numeric computations: sum/avg/min/max. Example: task='compute', input={'expression':'25*4'}",
      "usf_config":{"api_key": api_key}
    })
    writer = SubAgent({
      "id":"writer","name":"Writer","agent_type":"sub","context_mode":"ALWAYS_FULL",
      "description":"Write short summaries.",
      "usf_config":{"api_key": api_key}
    })
    mgr.add_sub_agent(calc, {"description": calc.description, "parameters":{"type":"object","properties":{"task":{"type":"string"}},"required":["task"]}}, alias="agent_calc")
    mgr.add_sub_agent(writer, {"description": writer.description, "parameters":{"type":"object","properties":{"task":{"type":"string"}},"required":["task"]}}, alias="agent_writer")

    # List tools exposed by manager (includes sub-agents as tools)
    tools = mgr.list_tools()
    print("Tools:", [t["function"]["name"] for t in tools])

    # Delegate a task explicitly
    res = await mgr.delegate("calc", {"task":"compute","input":{"expression":"12*7"}})
    print("Delegate:", res["success"], res.get("content",""))

asyncio.run(main())
# PY

Context shaping utilities

• sanitize_parent_context(msgs): keep only user and final-answer assistant messages for delegation
• shape_context_for_mode(mode, task, calling_agent_msgs=None, context_param=None) -> List[Message]

Example

from usf_agents.multi_agent.context import sanitize_parent_context, shape_context_for_mode
clean = sanitize_parent_context([
  {"role":"user","content":"Hi"},
  {"role":"assistant","content":"Plan ...","type":"agent_plan","plan":"..."},
  {"role":"assistant","content":"Final answer here"}
])
msgs = shape_context_for_mode("CONTEXT_PARAM", {"task":"write","input":{"topic":"X"}}, calling_agent_msgs=clean, context_param={"audience":"exec"})

Sequencing helpers

• validate_next_step(messages) -> raises if invalid to call run()
• run_until_final(agent, messages, tools, tool_router, max_loops=20) -> str

Example

from usf_agents.runtime.validate import validate_next_step
from usf_agents.runtime.safe_seq import run_until_final
# validate_next_step(messages) before calling agent.run()
# run_until_final(...) drives plan -> tool_calls -> tool results -> final

Graph Orchestration and Tracing

WorkflowGraph spec

• nodes: [{ id: str, type: 'agent'|'tool', ref: AgentId|tool_name, config?: Dict }]
• edges: [{ source: str, target: str, condition?: str }]

ExecutionEngine.run(entry_nodes: List[str], inputs: Dict[str, Any], max_steps=200) -> Dict[node_id, summary]

• Execute agents by calling wrapper.run_task(TaskPayload)
• Tool nodes may use an optional tool_executor callback

Tracing

• TraceRecorder: start(), record(event), end(status), snapshot()
• Visualization: to_mermaid(graph_spec, trace), to_graphviz(graph_spec, trace), to_json(trace)

Example

# run: USF_API_KEY=YOUR_KEY python - <<'PY'
import os, asyncio
from usf_agents.multi_agent.base import ManagerAgent, SubAgent
from usf_agents.graph.workflow import WorkflowGraph, ExecutionEngine
from usf_agents.multi_agent.registry import AgentRegistry
from usf_agents.trace.trace import TraceRecorder
from usf_agents.trace.visualize import to_mermaid, to_graphviz, to_json

async def main():
    api_key = os.getenv("USF_API_KEY")
    mgr = ManagerAgent({"id":"mgr","agent_type":"manager","usf_config":{"api_key": api_key}})
    sub = SubAgent({"id":"writer","agent_type":"sub","context_mode":"NONE","description":"Write short summaries","usf_config":{"api_key": api_key}})
    mgr.add_sub_agent(sub, {"description": sub.description, "parameters":{"type":"object","properties":{"task":{"type":"string"}},"required":["task"]}}, alias="agent_writer")

    reg = AgentRegistry(); reg.register(mgr); reg.register(sub)
    spec = {"nodes":[{"id":"start","type":"agent","ref":"mgr"}],"edges":[]}
    graph = WorkflowGraph(spec); recorder = TraceRecorder()
    engine = ExecutionEngine(graph, reg, recorder)
    outputs = await engine.run(entry_nodes=["start"], inputs={"start":{"task":"write","input":{"topic":"teamwork"}}})
    trace = recorder.snapshot()
    print("Mermaid:\\n", to_mermaid(spec, trace))
    print("Graphviz DOT:\\n", to_graphviz(spec, trace))
    print("Trace JSON:\\n", to_json(trace))

# asyncio.run(main())
# PY

---

Installation

pip install usf-agents

Requirements

• Python 3.9+
• USF API key: set environment variable USF_API_KEY

---

Quickstart (minimal and runnable)

All code below assumes your environment has USF_API_KEY set.

Windows PowerShell: $env:USF_API_KEY="YOUR_KEY"
Windows cmd: set USF_API_KEY=YOUR_KEY && python ...

1. Plain LLM (no tools)

# run: USF_API_KEY=YOUR_KEY python - <<'PY'
import os
import asyncio
from usf_agents import USFAgent

async def main():
    agent = USFAgent(
        {
            "api_key": os.getenv("USF_API_KEY"),
            "model": "usf-mini",
        }
    )
    async for chunk in agent.run("Hello, what's the capital of France?"):
        if chunk["type"] == "final_answer":
            print("Final:", chunk["content"])
            break

asyncio.run(main())
# PY

2. ManagerAgent, no tools (planning → final)

# run: USF_API_KEY=YOUR_KEY python - <<'PY'
import os
import asyncio
from usf_agents.multi_agent.base import ManagerAgent

async def main():
    mgr = ManagerAgent(
        {
            "id": "mgr",
            "name": "Manager",
            "agent_type": "manager",
            "usf_config": {
                "api_key": os.getenv("USF_API_KEY"),
                "model": "usf-mini",
            },
        }
    )
    async for chunk in mgr.usf.run("Summarize the benefits of unit testing."):
        if chunk["type"] == "final_answer":
            print("Final:", chunk["content"])
            break

asyncio.run(main())
# PY

Write good sub-agent descriptions

Provide a clear, scoped description on every SubAgent via the description field. This string is surfaced as the tool's function.description and helps the LLM pick the right tool when multiple tools are available.

Guidelines:

• Start with an action + domain: “Numeric computations…”, “Rapid web/knowledge lookup…”
• Clarify inputs/outputs briefly (align with task/input/context_param contract)
• Include one inline example matching the schema:
  • Example: task='compute', input={'expression': 'sum(prices)'}
• Avoid overlap between agents; make distinctions obvious and unambiguous

Bad:

• “General helper” or “Do analysis”

Good:

• “Generate/refactor code from specs with optional context_param snippets. Example: task='function', input={'signature': 'total_cost(prices: list[float]) -> float'}”

---

Tool Definitions Made Simple

Precedence for how a schema is determined (highest to lowest):

1. Schema passed to add_function_tool(..., schema=...)
2. Schema provided in the @tool decorator (schema=...)
3. Docstring parsing (YAML block → Google-style Args → NumPy-style Parameters)

Validation

• Always enforced: schema.required must equal the set of parameters with no default values in the function signature (excluding *args/**kwargs).
• Optional strict properties mode (strict=True) enforces exact equality between schema.properties keys and the function parameters.

A) Single function tool via Google-style docstring (no explicit schema)

# run: USF_API_KEY=YOUR_KEY python - <<'PY'
import os
import asyncio
from usf_agents.multi_agent.base import ManagerAgent

def calc(expression: str) -> int:
    """
    Evaluate a simple expression.
    Args:
        expression (str): A Python expression to evaluate.
    """
    return eval(expression)  # demo; use a safe evaluator in production

async def main():
    mgr = ManagerAgent(
        {
            "id": "mgr",
            "usf_config": {"api_key": os.getenv("USF_API_KEY"), "model": "usf-mini"},
        }
    )
    mgr.add_function_tool("calc", calc, alias="math_calc")
    final = await mgr.run_auto(
        [{"role": "user", "content": "Use math_calc to compute 25*4"}],
        mode="auto",
    )
    print("Final:", final)

asyncio.run(main())
# PY

B) NumPy-style docstring (no explicit schema)

# run: USF_API_KEY=YOUR_KEY python - <<'PY'
import os
import asyncio
from usf_agents.multi_agent.base import ManagerAgent

def greet(name: str) -> str:
    """
    Greet a user.

    Parameters
    ----------
    name : str
        Person to greet.
    """
    return f"Hello {name}!"

async def main():
    mgr = ManagerAgent(
        {
            "id": "mgr",
            "usf_config": {"api_key": os.getenv("USF_API_KEY"), "model": "usf-mini"},
        }
    )
    mgr.add_function_tool("greet", greet, alias="hello")
    final = await mgr.run_auto(
        [{"role": "user", "content": "Use hello for \"USF\""}],
        mode="auto",
    )
    print("Final:", final)

asyncio.run(main())
# PY

C) YAML block in docstring (takes precedence; no explicit schema)

# run: USF_API_KEY=YOUR_KEY python - <<'PY'
import os
import asyncio
from usf_agents.multi_agent.base import ManagerAgent

def http_get(url: str) -> dict:
    """
    Perform GET.

    ```yaml
    description: Simple HTTP GET (demo)
    parameters:
      type: object
      properties:
        url:
          type: string
          description: URL to fetch
      required: [url]
    ```
    """
    return {"status": 200, "body": "ok"}

async def main():
    mgr = ManagerAgent(
        {
            "id": "mgr",
            "usf_config": {"api_key": os.getenv("USF_API_KEY"), "model": "usf-mini"},
        }
    )
    mgr.add_function_tool("http_get", http_get)
    final = await mgr.run_auto(
        [{"role": "user", "content": "Call http_get with https://example.com"}],
        mode="auto",
    )
    print("Final:", final)

asyncio.run(main())
# PY

D) @tool decorator with defaults (name/alias/description) and docstring schema

# run: USF_API_KEY=YOUR_KEY python - <<'PY'
import os
import asyncio
from usf_agents.multi_agent.base import ManagerAgent
from usf_agents.runtime.decorators import tool

@tool(name="calc_sum", alias="sum_tool", description="Sum a list of integers")
def calc_sum(numbers: list[int]) -> int:
    """
    Sum integers.
    Args:
        numbers (list[int]): Values to add up.
    """
    return sum(numbers)

async def main():
    mgr = ManagerAgent(
        {
            "id": "mgr",
            "usf_config": {"api_key": os.getenv("USF_API_KEY"), "model": "usf-mini"},
        }
    )
    mgr.add_function_tool("calc_sum", calc_sum)  # decorator defaults used
    final = await mgr.run_auto(
        [{"role": "user", "content": "Use sum_tool to sum 10,20,30"}],
        mode="auto",
    )
    print("Final:", final)

asyncio.run(main())
# PY

E) @tool decorator with an explicit schema in the decorator (no explicit schema passed to add_function_tool)

# run: USF_API_KEY=YOUR_KEY python - <<'PY'
import os
import asyncio
from usf_agents.multi_agent.base import ManagerAgent
from usf_agents.runtime.decorators import tool

@tool(
    name="calc_sum",
    alias="sum_tool",
    description="Sum a list of integers",
    schema={
        "description": "Sum integers",
        "parameters": {
            "type": "object",
            "properties": {"numbers": {"type": "array", "description": "List of ints"}},
            "required": ["numbers"],
        },
    },
)
def calc_sum(numbers: list[int]) -> int:
    return sum(numbers)

async def main():
    mgr = ManagerAgent(
        {
            "id": "mgr",
            "usf_config": {"api_key": os.getenv("USF_API_KEY"), "model": "usf-mini"},
        }
    )
    # No schema passed here; decorator schema used and validated
    mgr.add_function_tool("calc_sum", calc_sum)
    final = await mgr.run_auto(
        [{"role": "user", "content": "Use sum_tool to sum 1..5"}],
        mode="auto",
    )
    print("Final:", final)

asyncio.run(main())
# PY

F) Explicit JSON schema passed to add_function_tool (overrides decorator/docstrings)

# run: USF_API_KEY=YOUR_KEY python - <<'PY'
import os
import asyncio
from usf_agents.multi_agent.base import ManagerAgent

def calc(expression: str) -> int:
    return eval(expression)

async def main():
    mgr = ManagerAgent(
        {
            "id": "mgr",
            "usf_config": {"api_key": os.getenv("USF_API_KEY"), "model": "usf-mini"},
        }
    )
    mgr.add_function_tool(
        "calc",
        calc,
        alias="math_calc",
        schema={
            "description": "Evaluate math expression",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {"type": "string", "description": "Python expression"}
                },
                "required": ["expression"],
            },
        },
        strict=False,  # set True to require properties set equality too
    )
    final = await mgr.run_auto(
        [{"role": "user", "content": "Use math_calc to compute 9*9"}],
        mode="auto",
    )
    print("Final:", final)

asyncio.run(main())
# PY

G) Type mapping example (string/number/boolean/object/array) via explicit schema

# run: USF_API_KEY=YOUR_KEY python - <<'PY'
import os
import asyncio
from usf_agents.multi_agent.base import ManagerAgent

def demo(a: str, n: int, flag: bool, cfg: dict, items: list) -> dict:
    return {"ok": True}

async def main():
    mgr = ManagerAgent(
        {
            "id": "mgr",
            "usf_config": {"api_key": os.getenv("USF_API_KEY"), "model": "usf-mini"},
        }
    )
    mgr.add_function_tool(
        "demo",
        demo,
        schema={
            "description": "Type mapping demo",
            "parameters": {
                "type": "object",
                "properties": {
                    "a": {"type": "string", "description": "string input"},
                    "n": {"type": "number", "description": "numeric input"},
                    "flag": {"type": "boolean", "description": "boolean toggle"},
                    "cfg": {"type": "object", "description": "config object"},
                    "items": {"type": "array", "description": "list of items"},
                },
                "required": ["a", "n", "flag", "cfg", "items"],
            },
        },
        strict=True,
    )
    final = await mgr.run_auto(
        [{"role": "user", "content": "Call demo with required fields"}],
        mode="auto",
    )
    print("Final:", final)

asyncio.run(main())
# PY

---

Auto Execution Modes

Summary

• disable: do not auto-run tools; return the assistant’s first tool_calls payload to the caller for manual handling.
• auto (default): auto-run both agent tools (sub-agents) and custom tools until final (or max_loops).
• agent: auto-run only agent tools; custom tool requests are returned to the caller as pending tool_calls.
• tool: auto-run only custom tools; agent tool requests are returned as pending tool_calls.

---

Multi-Agent in Fewer Lines

Add sub-agents in one line and run end-to-end.

# run: USF_API_KEY=YOUR_KEY python - <<'PY'
import os
import asyncio
from usf_agents.multi_agent.base import ManagerAgent

async def main():
    mgr = ManagerAgent(
        {
            "id": "mgr",
            "name": "Manager",
            "agent_type": "manager",
            "usf_config": {
                "api_key": os.getenv("USF_API_KEY"),
                "model": "usf-mini",
            },
        }
    )
    # Sub-agent appears as tool "agent_writer"
    mgr.add_sub_agent_simple(
        id="writer",
        alias="agent_writer",
        context_mode="NONE",
        description="Draft short outputs",
    )
    final = await mgr.run_auto(
        [{"role": "user", "content": "Ask agent_writer to write a haiku about teamwork."}],
        mode="auto",
    )
    print("Final:", final)

asyncio.run(main())
# PY

Production-grade pattern: explicit SubAgent instances (cleaner structure)

from usf_agents.multi_agent.base import SubAgent

api_key = os.getenv("USF_API_KEY")

# Model is optional; defaults to "usf-mini" if omitted
calculator = SubAgent(
    {
        "id": "calc",
        "name": "Calculator",
        "agent_type": "sub",
        "context_mode": "NONE",
        "description": "Perform numeric computations such as sum/avg/min/max. Example: expression='25*4'",
        "usf_config": {"api_key": api_key},
    }
)
researcher = SubAgent(
    {
        "id": "research",
        "name": "Researcher",
        "agent_type": "sub",
        "context_mode": "AGENT_DECIDED",
        "description": "Look up and synthesize knowledge from external sources. Example: topic='UltraWidget market trends'",
        "usf_config": {"api_key": api_key},
    }
)
coder = SubAgent(
    {
        "id": "coder",
        "name": "Code Assistant",
        "agent_type": "sub",
        "context_mode": "CONTEXT_PARAM",
        "description": "Generate or refactor code from specifications.",
        "usf_config": {"api_key": api_key},
    }
)
writer = SubAgent(
    {
        "id": "writer",
        "name": "Writer",
        "agent_type": "sub",
        "context_mode": "ALWAYS_FULL",
        "description": "Write polished summaries, briefs, or emails.",
        "usf_config": {"api_key": api_key},
    }
)

# Register any number in one call (varargs or list)
mgr.add_sub_agents(calculator, researcher, coder, writer)
# or
mgr.add_sub_agents([calculator, researcher])

Quick prototyping (dict specs) remains supported

mgr.add_sub_agents([
    {'id':'logs', 'alias':'agent_logs', 'context_mode':'AGENT_DECIDED', 'description':'AAnalyze system/application logs for errors and anomalies.'}
])

---

Registry Option (alternate flow)

Register tools with ToolRegistry and use run_auto façade.

# run: USF_API_KEY=YOUR_KEY python - <<'PY'
import os
import asyncio
from usf_agents import USFAgent
from usf_agents.runtime.tool_registry import ToolRegistry
from usf_agents.runtime.auto_exec import run_auto

def calc(expression: str) -> int:
    return eval(expression)

async def main():
    agent = USFAgent(
        {
            "api_key": os.getenv("USF_API_KEY"),
            "model": "usf-mini",
        }
    )
    registry = ToolRegistry()
    registry.register_function(
        name="calc",
        func=calc,
        schema={
            "description": "calc",
            "parameters": {
                "type": "object",
                "properties": {"expression": {"type": "string"}},
                "required": ["expression"],
            },
        },
        examples=[{"name": "smoke", "args": {"expression": "2+3"}, "expect": 5}],
    )
    final = await run_auto(
        agent,
        [{"role": "user", "content": "Use calc to compute 25*4"}],
        registry=registry,
        mode="auto",
    )
    print("Final:", final)

asyncio.run(main())
# PY

---

Advanced Cookbook (end-to-end)

A) Incident Response Orchestrator (compressed)

# Pseudo-runnable: fill USF_API_KEY and run in a cell/script
import os
import asyncio
from usf_agents.multi_agent.base import ManagerAgent, SubAgent

async def main():
    api_key = os.getenv("USF_API_KEY")
    mgr = ManagerAgent(
        {
            "id": "ir_mgr",
            "name": "Incident Orchestrator",
            "agent_type": "manager",
            "usf_config": {"api_key": api_key},
        }
    )

    logs = SubAgent(
        {
            "id": "logs",
            "name": "Logs Analyzer",
            "agent_type": "sub",
            "context_mode": "AGENT_DECIDED",
            "description": "Analyze system/application logs for errors and anomalies.",
            "usf_config": {"api_key": api_key},
        }
    )
    rootcause = SubAgent(
        {
            "id": "rootcause",
            "name": "Root Cause",
            "agent_type": "sub",
            "context_mode": "ALWAYS_FULL",
            "description": "Infer root cause from logs, metrics, and incident context.",
            "usf_config": {"api_key": api_key},
        }
    )
    remediate = SubAgent(
        {
            "id": "remediate",
            "name": "Remediate",
            "agent_type": "sub",
            "context_mode": "CONTEXT_PARAM",
            "description": "Suggest remediation steps for known issues.",
            "usf_config": {"api_key": api_key},
        }
    )
    comms = SubAgent(
        {
            "id": "comms",
            "name": "Comms",
            "agent_type": "sub",
            "context_mode": "ALWAYS_FULL",
            "description": "Draft stakeholder communications (incident reports, updates, notices).",
            "usf_config": {"api_key": api_key},
        }
    )

    mgr.add_sub_agents(logs, rootcause, remediate, comms)

    brief = (
        "SEV-1 Incident: summarize logs, root cause, remediations, "
        "and draft stakeholder update."
    )
    final = await mgr.run_auto([{"role": "user", "content": brief}], mode="auto")
    print("Final:", final)

# asyncio.run(main())

B) Product Launch Orchestrator (compressed)

import os
import asyncio
from usf_agents.multi_agent.base import ManagerAgent, SubAgent

async def main():
    api_key = os.getenv("USF_API_KEY")
    mgr = ManagerAgent(
        {
            "id": "pl_mgr",
            "name": "Launch Orchestrator",
            "agent_type": "manager",
            "usf_config": {"api_key": api_key},
        }
    )

    req = SubAgent(
        {
            "id": "requirements",
            "name": "Requirements",
            "agent_type": "sub",
            "context_mode": "ALWAYS_FULL",
            "description": "Extract and clarify project or product requirements.",
            "usf_config": {"api_key": api_key},
        }
    )
    planner = SubAgent(
        {
            "id": "planner",
            "name": "Planner",
            "agent_type": "sub",
            "context_mode": "NONE",
            "description": "Plan project milestones, timelines, or resources.",
            "usf_config": {"api_key": api_key},
        }
    )
    risk = SubAgent(
        {
            "id": "risk",
            "name": "Risk",
            "agent_type": "sub",
            "context_mode": "AGENT_DECIDED",
            "description": "Identify risks and propose mitigations for projects.",
            "usf_config": {"api_key": api_key},
        }
    )
    writer = SubAgent(
        {
            "id": "writer",
            "name": "Writer",
            "agent_type": "sub",
            "context_mode": "ALWAYS_FULL",
            "description": "Produce clear summaries, briefs, and reports for stakeholders.",
            "usf_config": {"api_key": api_key},
        }
    )
    mgr.add_sub_agents(req, planner, risk, writer)

    ask = (
        "Extract constraints, plan timeline/cost, risks/mitigation, "
        "and craft launch email + internal brief."
    )
    final = await mgr.run_auto([{"role": "user", "content": ask}], mode="auto")
    print("Final:", final)

C) Data Science Orchestrator (compressed)

import os
import asyncio
from usf_agents.multi_agent.base import ManagerAgent, SubAgent

async def main():
    api_key = os.getenv("USF_API_KEY")
    mgr = ManagerAgent(
        {
            "id": "ds_mgr",
            "name": "DS Orchestrator",
            "agent_type": "manager",
            "usf_config": {"api_key": api_key},
        }
    )

    fetch = SubAgent(
        {
            "id": "fetch",
            "name": "Data Fetcher",
            "agent_type": "sub",
            "context_mode": "CONTEXT_PARAM",
            "description": "Fetch datasets from sources given a query.",
            "usf_config": {"api_key": api_key},
        }
    )
    eda = SubAgent(
        {
            "id": "eda",
            "name": "EDA Analyzer",
            "agent_type": "sub",
            "context_mode": "AGENT_DECIDED",
            "description": "Perform exploratory data analysis on datasets.",
            "usf_config": {"api_key": api_key},
        }
    )
    model = SubAgent(
        {
            "id": "model",
            "name": "Model Coder",
            "agent_type": "sub",
            "context_mode": "CONTEXT_PARAM",
            "description": "Generate or refine ML model training and inference code.",
            "usf_config": {"api_key": api_key},
        }
    )
    report = SubAgent(
        {
            "id": "report",
            "name": "Report Writer",
            "agent_type": "sub",
            "context_mode": "ALWAYS_FULL",
            "description": "Write ML model training/inference code.",
            "usf_config": {"api_key": api_key},
        }
    )
    mgr.add_sub_agents(fetch, eda, model, report)

    req = "Churn dataset prototype: fetch, EDA, baseline code, concise report."
    final = await mgr.run_auto([{"role": "user", "content": req}], mode="auto")
    print("Final:", final)

D) Manager Dynamic Orchestrator (compressed)

import os
import asyncio
from usf_agents.multi_agent.base import ManagerAgent, SubAgent

async def main():
    api_key = os.getenv("USF_API_KEY")
    mgr = ManagerAgent(
        {
            "id": "mgr",
            "name": "Dynamic Orchestrator",
            "agent_type": "manager",
            "usf_config": {"api_key": api_key},
        }
    )

    calculator = SubAgent(
        {
            "id": "calc",
            "name": "Calculator",
            "agent_type": "sub",
            "context_mode": "NONE",
            "description": "Numeric computations (sum/avg/min/max).}",
            "usf_config": {"api_key": api_key},
        }
    )
    researcher = SubAgent(
        {
            "id": "research",
            "name": "Researcher",
            "agent_type": "sub",
            "context_mode": "AGENT_DECIDED",
            "description": "Rapid web/knowledge lookup and synthesis.",
            "usf_config": {"api_key": api_key},
        }
    )
    coder = SubAgent(
        {
            "id": "coder",
            "name": "Code Assistant",
            "agent_type": "sub",
            "context_mode": "CONTEXT_PARAM",
            "description": "Generate/refactor code from specs with optional context_param snippets.",
            "usf_config": {"api_key": api_key},
        }
    )
    writer = SubAgent(
        {
            "id": "writer",
            "name": "Writer",
            "agent_type": "sub",
            "context_mode": "ALWAYS_FULL",
            "description": "Executive summaries, briefs, and emails.",
            "usf_config": {"api_key": api_key},
        }
    )
    mgr.add_sub_agents(calculator, researcher, coder, writer)

    task = (
        "For 'UltraWidget': compute total cost, research trends, generate Python function "
        "total_cost(prices: list[float]) -> float, and write an executive summary."
    )
    final = await mgr.run_auto([{"role": "user", "content": task}], mode="auto")
    print("Final:", final)

---

Troubleshooting / FAQ

• Error: “no explicit schema and no parseable docstring”
  • Add a Google/NumPy docstring or a YAML block, or pass schema explicitly to add_function_tool, or provide schema in the @tool decorator.
• Error: “required mismatch”
  • Ensure your schema.required equals the set of function parameters with no default values.
• Error with strict=True: “properties mismatch”
  • Ensure schema.parameters.properties keys match your function parameters exactly.
• YAML parsing
  • YAML blocks are optional. If PyYAML isn’t installed, YAML parsing is skipped and Google/NumPy parsing is attempted automatically.
• Tool name collisions
  • Per-agent tool names must be unique across native tools and sub-agents. Use aliases to disambiguate.

---

Contributing / License

License: USF Agents SDK License
Copyright (c) 2025 UltraSafe AI Team

PERMITTED USE:

• Anyone may use this software for any purpose

RESTRICTED ACTIVITIES:

• No one may modify the code
• No one may use the code for commercial purposes
• No one may use the code to create competitive products

ATTRIBUTION:

• All copies of this software must retain this license notice
• Any use of this software must include attribution to UltraSafe AI Team