jeevesagent.team¶

Team — sibling-style builders for multi-agent architectures.

Coming from LangGraph, CrewAI, AutoGen, or the OpenAI Agents SDK, you’d expect to construct a multi-agent team like this:

team = create_supervisor([researcher, writer], model="gpt-4")
result = await team.invoke(...)

In JeevesAgent the same shape is Team:

team = Team.supervisor(
    workers={"researcher": researcher, "writer": writer},
    instructions="manage the pipeline",
    model="gpt-4.1-mini",
)
result = await team.run("write me a blog post")

Under the hood Team returns a regular Agent — exactly what Agent(architecture=Supervisor(...)) would produce. The two shapes are interchangeable; Team is the “familiar-from-other-frameworks” facade for the common case, while the nested Agent(architecture=...) form remains the path for recursive composition (wrapping a Supervisor in Reflexion, etc.).

Provided builders¶

Team.supervisor() — coordinator + workers
Team.swarm() — peer agents handing off control
Team.router() — classify-and-dispatch
Team.debate() — N debaters + optional judge
Team.actor_critic() — actor + critic pair
Team.blackboard() — shared workspace + coordinator

Each method exposes every Agent constructor kwarg explicitly (rather than forwarding through **kwargs) so that IDEs and type-checkers surface the full parameter list with proper types, defaults, and docstrings on hover.

Standalone-run helper¶

run_architecture() builds a minimal Agent shell around any Architecture instance and runs it. Useful for testing orchestrators in isolation without wiring a full Agent yourself:

sup = Supervisor(workers={"a": agent_a})
result = await run_architecture(sup, "do the thing", model="gpt-4.1-mini")

Classes¶

Team

Namespace for multi-agent team builders.

Functions¶

run_architecture(→ jeevesagent.core.types.RunResult)

Run an Architecture once with a minimal Agent shell.

Module Contents¶

class jeevesagent.team.Team[source]¶

Namespace for multi-agent team builders.

Every classmethod returns a fully-built Agent whose architecture is the corresponding multi-agent strategy. The returned Agent has the standard run / stream / etc. interface — call sites don’t change between single-agent and team agents.

static actor_critic(actor: jeevesagent.agent.api.Agent, critic: jeevesagent.agent.api.Agent, *, instructions: str = '', model: jeevesagent.core.protocols.Model | str | None = None, memory: jeevesagent.core.protocols.Memory | None = None, runtime: jeevesagent.core.protocols.Runtime | None = None, budget: jeevesagent.core.protocols.Budget | None = None, permissions: jeevesagent.core.protocols.Permissions | None = None, hooks: jeevesagent.security.hooks.HookRegistry | None = None, tools: list[jeevesagent.tools.registry.Tool | collections.abc.Callable[Ellipsis, object]] | jeevesagent.core.protocols.ToolHost | jeevesagent.tools.registry.Tool | collections.abc.Callable[Ellipsis, object] | None = None, telemetry: jeevesagent.core.protocols.Telemetry | None = None, audit_log: jeevesagent.security.audit.AuditLog | None = None, max_turns: int = DEFAULT_MAX_TURNS, auto_consolidate: bool = False, skills: list[Any] | None = None, max_rounds: int = 3, approval_threshold: float = 0.9, critique_template: str | None = None, refine_template: str | None = None) → jeevesagent.agent.api.Agent[source]¶: Build an actor-critic pair where the critic reviews the actor’s output (with structured JSON scoring + rubric) and the actor refines below approval_threshold.

static blackboard(agents: dict[str, jeevesagent.agent.api.Agent], *, coordinator: jeevesagent.agent.api.Agent | None = None, decider: jeevesagent.agent.api.Agent | None = None, instructions: str = '', model: jeevesagent.core.protocols.Model | str | None = None, memory: jeevesagent.core.protocols.Memory | None = None, runtime: jeevesagent.core.protocols.Runtime | None = None, budget: jeevesagent.core.protocols.Budget | None = None, permissions: jeevesagent.core.protocols.Permissions | None = None, hooks: jeevesagent.security.hooks.HookRegistry | None = None, tools: list[jeevesagent.tools.registry.Tool | collections.abc.Callable[Ellipsis, object]] | jeevesagent.core.protocols.ToolHost | jeevesagent.tools.registry.Tool | collections.abc.Callable[Ellipsis, object] | None = None, telemetry: jeevesagent.core.protocols.Telemetry | None = None, audit_log: jeevesagent.security.audit.AuditLog | None = None, max_turns: int = DEFAULT_MAX_TURNS, auto_consolidate: bool = False, skills: list[Any] | None = None, max_rounds: int = 10, coordinator_instructions: str | None = None, decider_instructions: str | None = None) → jeevesagent.agent.api.Agent[source]¶: Build a blackboard team where agents collaborate via a shared workspace; an optional coordinator selects who acts each round and an optional decider decides when the work is done.

static debate(debaters: list[jeevesagent.agent.api.Agent], *, judge: jeevesagent.agent.api.Agent | None = None, instructions: str = '', model: jeevesagent.core.protocols.Model | str | None = None, memory: jeevesagent.core.protocols.Memory | None = None, runtime: jeevesagent.core.protocols.Runtime | None = None, budget: jeevesagent.core.protocols.Budget | None = None, permissions: jeevesagent.core.protocols.Permissions | None = None, hooks: jeevesagent.security.hooks.HookRegistry | None = None, tools: list[jeevesagent.tools.registry.Tool | collections.abc.Callable[Ellipsis, object]] | jeevesagent.core.protocols.ToolHost | jeevesagent.tools.registry.Tool | collections.abc.Callable[Ellipsis, object] | None = None, telemetry: jeevesagent.core.protocols.Telemetry | None = None, audit_log: jeevesagent.security.audit.AuditLog | None = None, max_turns: int = DEFAULT_MAX_TURNS, auto_consolidate: bool = False, skills: list[Any] | None = None, rounds: int = 2, convergence_check: bool = True, convergence_similarity: float = 0.85, debater_instructions: str | None = None, judge_instructions: str | None = None) → jeevesagent.agent.api.Agent[source]¶: Build a multi-agent debate where debaters argue for rounds (with optional convergence early-exit). If judge is provided, the judge synthesizes a final answer; otherwise majority vote wins.

static router(routes: list[jeevesagent.architecture.RouterRoute], *, instructions: str = '', model: jeevesagent.core.protocols.Model | str | None = None, memory: jeevesagent.core.protocols.Memory | None = None, runtime: jeevesagent.core.protocols.Runtime | None = None, budget: jeevesagent.core.protocols.Budget | None = None, permissions: jeevesagent.core.protocols.Permissions | None = None, hooks: jeevesagent.security.hooks.HookRegistry | None = None, tools: list[jeevesagent.tools.registry.Tool | collections.abc.Callable[Ellipsis, object]] | jeevesagent.core.protocols.ToolHost | jeevesagent.tools.registry.Tool | collections.abc.Callable[Ellipsis, object] | None = None, telemetry: jeevesagent.core.protocols.Telemetry | None = None, audit_log: jeevesagent.security.audit.AuditLog | None = None, max_turns: int = DEFAULT_MAX_TURNS, auto_consolidate: bool = False, skills: list[Any] | None = None, fallback_route: str | None = None, require_confidence_above: float = 0.0, classifier_prompt: str | None = None) → jeevesagent.agent.api.Agent[source]¶: Build a router that classifies once and dispatches to ONE specialist Agent. Cheaper than Supervisor for tasks with clear specialist boundaries (one classifier call + one specialist run, no synthesis pass).

static supervisor(workers: dict[str, jeevesagent.agent.api.Agent], *, instructions: str = '', model: jeevesagent.core.protocols.Model | str | None = None, memory: jeevesagent.core.protocols.Memory | None = None, runtime: jeevesagent.core.protocols.Runtime | None = None, budget: jeevesagent.core.protocols.Budget | None = None, permissions: jeevesagent.core.protocols.Permissions | None = None, hooks: jeevesagent.security.hooks.HookRegistry | None = None, tools: list[jeevesagent.tools.registry.Tool | collections.abc.Callable[Ellipsis, object]] | jeevesagent.core.protocols.ToolHost | jeevesagent.tools.registry.Tool | collections.abc.Callable[Ellipsis, object] | None = None, telemetry: jeevesagent.core.protocols.Telemetry | None = None, audit_log: jeevesagent.security.audit.AuditLog | None = None, max_turns: int = DEFAULT_MAX_TURNS, auto_consolidate: bool = False, skills: list[Any] | None = None, instructions_template: str | None = None, delegate_tool_name: str = 'delegate', forward_tool_name: str = 'forward_message') → jeevesagent.agent.api.Agent[source]¶

Build a coordinator Agent that delegates to workers.

The coordinator can call delegate(worker, instructions) to dispatch a subtask, or forward_message(worker) to return a worker’s output verbatim. Multiple delegations in one turn run in parallel.

static swarm(agents: dict[str, jeevesagent.agent.api.Agent | jeevesagent.architecture.swarm.Handoff], entry_agent: str, *, instructions: str = '', model: jeevesagent.core.protocols.Model | str | None = None, memory: jeevesagent.core.protocols.Memory | None = None, runtime: jeevesagent.core.protocols.Runtime | None = None, budget: jeevesagent.core.protocols.Budget | None = None, permissions: jeevesagent.core.protocols.Permissions | None = None, hooks: jeevesagent.security.hooks.HookRegistry | None = None, tools: list[jeevesagent.tools.registry.Tool | collections.abc.Callable[Ellipsis, object]] | jeevesagent.core.protocols.ToolHost | jeevesagent.tools.registry.Tool | collections.abc.Callable[Ellipsis, object] | None = None, telemetry: jeevesagent.core.protocols.Telemetry | None = None, audit_log: jeevesagent.security.audit.AuditLog | None = None, max_turns: int = DEFAULT_MAX_TURNS, auto_consolidate: bool = False, skills: list[Any] | None = None, max_handoffs: int = 8, detect_cycles: bool = True, pass_full_history: bool = True, handoff_tool_name: str = 'handoff') → jeevesagent.agent.api.Agent[source]¶

Build a peer-swarm of agents that hand off control via a handoff tool (or per-target transfer_to_<name> tools when peers are wrapped in Handoff with an input_type).

entry_agent is the peer that receives the first message.

async jeevesagent.team.run_architecture(architecture: jeevesagent.architecture.Architecture, prompt: str, *, instructions: str = '', model: jeevesagent.core.protocols.Model | str | None = None, memory: jeevesagent.core.protocols.Memory | None = None, runtime: jeevesagent.core.protocols.Runtime | None = None, budget: jeevesagent.core.protocols.Budget | None = None, permissions: jeevesagent.core.protocols.Permissions | None = None, hooks: jeevesagent.security.hooks.HookRegistry | None = None, tools: list[jeevesagent.tools.registry.Tool | collections.abc.Callable[Ellipsis, object]] | jeevesagent.core.protocols.ToolHost | jeevesagent.tools.registry.Tool | collections.abc.Callable[Ellipsis, object] | None = None, telemetry: jeevesagent.core.protocols.Telemetry | None = None, audit_log: jeevesagent.security.audit.AuditLog | None = None, max_turns: int = DEFAULT_MAX_TURNS, auto_consolidate: bool = False) → jeevesagent.core.types.RunResult[source]¶

Run an Architecture once with a minimal Agent shell.

Useful for testing orchestrators in isolation or for one-shot scripts where you don’t want to construct an Agent yourself.

The default model is the framework’s resolver default (set via model= or env / config); pass an explicit model or string id to override.

Example:

sup = Supervisor(workers={"a": agent_a})
result = await run_architecture(
    sup, "do the thing", model="gpt-4.1-mini"
)