Autonomy API

autonomy

Runtime agent spawning and tool discovery.

The autonomy module enables the orchestrator to design and spawn new agents at runtime. An LLM produces an AgentSpec (structured output describing an agent), and the AgentFactory hydrates it into a live BaseAgent. The ToolRegistry provides discoverable tools.

AgentFactory

AgentFactory(
    registry: ToolRegistry | None = None,
    default_model: str = "openai:gpt-4.1",
    api_key: str = "",
    *,
    tool_factory: DynamicToolFactory | None = None,
)

Creates live agents from AgentSpec definitions.

The factory resolves tools from the ToolRegistry and constructs Pydantic output models from JSON Schema at runtime.

Initialize with an optional registry, default model, and API key.

Parameters:

Name	Type	Description	Default
registry	ToolRegistry | None	Optional :class:ToolRegistry for resolving :class:ToolSpec references. Created empty if omitted.	None
default_model	str	Fallback model when spec.model is empty.	'openai:gpt-4.1'
api_key	str	API key used when spawning agents from a model string.	''
tool_factory	DynamicToolFactory | None	Optional :class:DynamicToolFactory for materializing :class:GeneratedToolSpec entries inside AgentSpec.tools. When None, generated tool specs are logged + dropped (matches the existing graceful-skip behavior for unknown registry names).	None

Source code in orqest/autonomy/factory.py

def __init__(
    self,
    registry: ToolRegistry | None = None,
    default_model: str = "openai:gpt-4.1",
    api_key: str = "",
    *,
    tool_factory: DynamicToolFactory | None = None,
) -> None:
    """Initialize with an optional registry, default model, and API key.

    Args:
        registry: Optional :class:`ToolRegistry` for resolving
            :class:`ToolSpec` references. Created empty if omitted.
        default_model: Fallback model when ``spec.model`` is empty.
        api_key: API key used when spawning agents from a model string.
        tool_factory: Optional :class:`DynamicToolFactory` for
            materializing :class:`GeneratedToolSpec` entries inside
            ``AgentSpec.tools``. When ``None``, generated tool specs
            are logged + dropped (matches the existing graceful-skip
            behavior for unknown registry names).

    """
    self._registry = registry or ToolRegistry()
    self._default_model = default_model
    self._api_key = api_key
    self._tool_factory = tool_factory

spawn

spawn(
    spec: AgentSpec, *, model: Model | None = None
) -> DynamicAgent

Hydrate an AgentSpec into a runnable DynamicAgent.

Parameters:

Name	Type	Description	Default
spec	AgentSpec	The agent specification to hydrate. spec.tools may mix :class:ToolSpec (registry lookup) and :class:GeneratedToolSpec (sandbox-spawned). Generated tools require a tool_factory configured at construction.	required
model	Model | None	Optional pre-built Model instance. When provided, the factory uses it directly instead of resolving from spec.model. This is the recommended way to inject TestModel in tests.	None

Source code in orqest/autonomy/factory.py

def spawn(self, spec: AgentSpec, *, model: Model | None = None) -> DynamicAgent:
    """Hydrate an AgentSpec into a runnable DynamicAgent.

    Args:
        spec: The agent specification to hydrate. ``spec.tools`` may
            mix :class:`ToolSpec` (registry lookup) and
            :class:`GeneratedToolSpec` (sandbox-spawned). Generated
            tools require a ``tool_factory`` configured at construction.
        model: Optional pre-built Model instance. When provided, the
            factory uses it directly instead of resolving from spec.model.
            This is the recommended way to inject TestModel in tests.

    """
    # Dispatch on whichever output-shape declaration the spec carries.
    # The AgentSpec validator already enforced exactly-one-of, so we can
    # branch cleanly here.
    if spec.output_type is not None:
        output_type = spec.output_type
    else:
        # JSON Schema path — synthesise a Pydantic model at runtime.
        assert spec.output_schema is not None  # validator guarantee
        output_type = self._schema_to_model(spec.name, spec.output_schema)
    # Pass agent_id (= spec.name) so Tier-2 sandbox routes execution into
    # the agent's per-agent subfolder + .venv. Tier-0 / Tier-1 ignore.
    tools = self._resolve_tools(spec.tools, agent_id=spec.name)

    prompt = spec.system_prompt
    if spec.constraints:
        constraint_text = "\n".join(f"- {c}" for c in spec.constraints)
        prompt += f"\n\nConstraints (you MUST follow these):\n{constraint_text}"

    if model is not None:
        return DynamicAgent(
            agent_name=spec.name,
            system_prompt=prompt,
            output_type=output_type,
            model=model,
            tools=tools if tools else None,
        )

    return DynamicAgent(
        agent_name=spec.name,
        system_prompt=prompt,
        output_type=output_type,
        model=spec.model or self._default_model,
        api_key=self._api_key,
        tools=tools if tools else None,
    )

DynamicAgent

DynamicAgent(
    agent_name: str,
    system_prompt: str,
    output_type: type[OutputT],
    *,
    model: Model | str,
    api_key: str | None = None,
    retries: int = 3,
    tools: list[Tool] | None = None,
    toolsets: list[Any] | None = None,
    truncated_history: int = 100,
    history_processors: list | None = None,
    result_budget: int | None = 20000,
    context_manager: ContextManager | None = None,
    model_settings: ModelSettings | None = None,
    reasoning: ReasoningEffort | None = None,
    confidence_protocol: Any = None,
)

An agent created at runtime from an AgentSpec.

Unlike user-defined agents, DynamicAgent's _run_implementation is generic: it takes the latest user message and calls call_model.

Source code in orqest/agents/base_agent.py

def __init__(
    self,
    agent_name: str,
    system_prompt: str,
    output_type: type[OutputT],
    *,
    model: Model | str,
    api_key: str | None = None,
    retries: int = 3,
    tools: list[Tool] | None = None,
    toolsets: list[Any] | None = None,
    truncated_history: int = 100,
    history_processors: list | None = None,
    result_budget: int | None = 20_000,
    context_manager: ContextManager | None = None,
    model_settings: ModelSettings | None = None,
    reasoning: ReasoningEffort | None = None,
    confidence_protocol: Any = None,
):
    """Initialize the agent.

    Args:
        agent_name: Name for logging and identification.
        system_prompt: System prompt guiding agent behavior.
        output_type: Pydantic model class for structured output.
        model: A pydantic-ai Model instance, or a 'provider:model_id' string.
        api_key: Required when model is a string; passed to the provider.
        retries: Retry attempts for failed LLM calls.
        tools: Individual Tool instances to register.
        toolsets: Toolset objects providing collections of tools.
        truncated_history: Max recent messages kept by the default history processor.
        history_processors: Custom processors; defaults to keep_recent_messages.
        result_budget: Max chars for tool result content before truncation.
            None disables budgeting. Default 20,000.
        context_manager: Optional ContextManager for token-aware compaction.
            When present, its compact method is prepended as the first
            history processor.
        model_settings: Optional pydantic-ai ``ModelSettings`` applied to
            every model call (e.g. ``ModelSettings(temperature=0.0,
            seed=42)``). Default ``None`` preserves provider defaults.
        reasoning: Optional provider-agnostic reasoning/thinking effort —
            one of ``'minimal'`` | ``'low'`` | ``'medium'`` | ``'high'``.
            Translated to the right provider-specific ``ModelSettings``
            key (``anthropic_thinking`` / ``openai_reasoning_effort`` /
            ``google_thinking_config`` / ``openrouter_reasoning``) and
            merged into ``model_settings`` — explicit ``model_settings``
            keys win on conflict. The model's provider must be one
            orqest can resolve, or construction raises ``ValueError``.
        confidence_protocol: Optional agent-level default
            ``ConfidenceProtocol`` used by :meth:`run_enriched`.
            ``None`` means ``run_enriched`` returns a bare
            ``EnrichedOutput`` unless a protocol is passed per call.

    """
    if isinstance(model, str):
        if api_key is None:
            raise ValueError(
                "api_key is required when model is a string name. "
                "Pass api_key explicitly or provide a Model instance instead."
            )
        self._model = resolve_model(model, api_key=api_key)
    elif isinstance(model, Model):
        self._model = model
    else:
        raise TypeError(
            f"model must be a pydantic-ai Model instance or a 'provider:model_id' string, "
            f"got {type(model).__name__}"
        )

    self.reasoning = reasoning
    if reasoning is not None:
        provider = model if isinstance(model, str) else self._model.system
        reasoning_settings = resolve_reasoning_settings(
            provider, reasoning, base=model_settings
        )
        # Reasoning fills in; explicit model_settings keys win on conflict.
        model_settings = {**reasoning_settings, **(model_settings or {})}

    self.agent_name = agent_name
    self.system_prompt = system_prompt
    _validate_output_type_for_llm_structured_output(agent_name, output_type)
    self.output_type = output_type
    self.retries = retries
    self.truncated_history = truncated_history
    self.model_settings = model_settings

    self.tools: list[Tool] = [
        t if isinstance(t, Tool) else Tool(t) for t in (tools or [])
    ]
    self.toolsets = list(toolsets) if toolsets else []

    if history_processors is not None:
        self._history_processors = list(history_processors)
    else:
        self._history_processors = [
            partial(keep_recent_messages, max_messages=truncated_history)
        ]

    if result_budget is not None:
        self._history_processors.insert(
            0, partial(budget_tool_results, max_result_chars=result_budget)
        )

    if context_manager is not None:
        self._history_processors.insert(0, context_manager.compact)

    # ALWAYS run an orphan-tool-return sweep last. Any earlier processor
    # (custom or built-in) can split a tool_call/tool_return pair across
    # the trim boundary; both the OpenAI chat/completions transport AND
    # the Responses transport reject orphan tool returns with a 400.
    # See :func:`orqest.agents.context_manager._repair_orphan_tool_returns`.
    from orqest.agents.context_manager import _repair_orphan_tool_returns
    self._history_processors.append(_repair_orphan_tool_returns)

    self._confidence_protocol = confidence_protocol
    self._agent: Agent | None = None

ExecutionResult

Result of executing the full goal.

MetaOrchestrator

MetaOrchestrator(
    planner: BaseAgent,
    factory: Any,
    registry: Any,
    *,
    memory: Any | None = None,
    hooks: HookRunner | None = None,
    max_subtasks: int = 10,
    max_spawn_depth: int = 3,
    metacognition: Any = None,
    bus: EventBus | None = None,
)

Orchestrator that decomposes goals and spawns agents as needed.

The MetaOrchestrator: 1. Takes a high-level goal from the user 2. Uses a planner agent to decompose it into subtasks 3. For each subtask, finds an existing agent or spawns a new one 4. Executes the subtasks sequentially (v1) 5. Collects results and produces a summary 6. Persists each spawned agent spec to memory (at spawn time) for reuse

Initialize the MetaOrchestrator.

Parameters:

Name	Type	Description	Default
planner	BaseAgent	Agent that decomposes goals into TaskDecomposition.	required
factory	Any	AgentFactory that spawns agents from AgentSpec.	required
registry	Any	ToolRegistry for tool lookup.	required
memory	Any | None	Optional MemoryStore for persisting learned specs.	None
hooks	HookRunner | None	Optional HookRunner for lifecycle events.	None
max_subtasks	int	Upper bound on subtasks per goal.	10
max_spawn_depth	int	Maximum nesting depth for spawned agents.	3
metacognition	Any	Optional :class:MetacognitionConfig. When set, after each successful subtask the orchestrator inspects the result's confidence; if it falls below redecompose_threshold (and the re-decomposition budget remains), the planner is re-invoked to rewrite the remaining subtasks. None preserves the legacy straight-through behavior.	None
bus	EventBus | None	Optional :class:~orqest.observability.EventBus. When set, the orchestrator emits a typed metacognition.redecomposition_triggered event each time low confidence triggers re-decomposition.	None

Source code in orqest/autonomy/meta.py

def __init__(
    self,
    planner: BaseAgent,
    factory: Any,
    registry: Any,
    *,
    memory: Any | None = None,
    hooks: HookRunner | None = None,
    max_subtasks: int = 10,
    max_spawn_depth: int = 3,
    metacognition: Any = None,
    bus: EventBus | None = None,
) -> None:
    """Initialize the MetaOrchestrator.

    Args:
        planner: Agent that decomposes goals into TaskDecomposition.
        factory: AgentFactory that spawns agents from AgentSpec.
        registry: ToolRegistry for tool lookup.
        memory: Optional MemoryStore for persisting learned specs.
        hooks: Optional HookRunner for lifecycle events.
        max_subtasks: Upper bound on subtasks per goal.
        max_spawn_depth: Maximum nesting depth for spawned agents.
        metacognition: Optional :class:`MetacognitionConfig`. When
            set, after each successful subtask the orchestrator
            inspects the result's confidence; if it falls below
            ``redecompose_threshold`` (and the re-decomposition
            budget remains), the planner is re-invoked to rewrite
            the remaining subtasks. ``None`` preserves the legacy
            straight-through behavior.
        bus: Optional :class:`~orqest.observability.EventBus`. When
            set, the orchestrator emits a typed
            ``metacognition.redecomposition_triggered`` event each
            time low confidence triggers re-decomposition.

    """
    self._planner = planner
    self._factory = factory
    self._registry = registry
    self._memory = memory
    self._hooks = hooks or HookRunner()
    self._max_subtasks = max_subtasks
    self._max_spawn_depth = max_spawn_depth
    self._metacognition = metacognition
    self._bus = bus
    self._spawned_agents: dict[str, BaseAgent] = {}

spawned_agents property

spawned_agents: dict[str, BaseAgent]

Access the cache of dynamically spawned agents.

solve async

solve(goal: str) -> ExecutionResult

Decompose a goal into subtasks, spawn agents, and execute.

When :class:MetacognitionConfig is configured, low-confidence subtask results trigger re-decomposition of the remaining subtasks (bounded by max_redecompositions).

Source code in orqest/autonomy/meta.py

async def solve(self, goal: str) -> ExecutionResult:
    """Decompose a goal into subtasks, spawn agents, and execute.

    When :class:`MetacognitionConfig` is configured, low-confidence
    subtask results trigger re-decomposition of the remaining
    subtasks (bounded by ``max_redecompositions``).
    """
    start = time.monotonic()

    decomposition = await self._decompose(goal)
    subtasks = list(decomposition.subtasks[: self._max_subtasks])

    results: list[SubTaskResult] = []
    context: dict[str, Any] = {}
    redecomposition_count = 0

    i = 0
    while i < len(subtasks):
        subtask = subtasks[i]
        result = await self._execute_subtask(subtask, context)
        results.append(result)
        if result.success:
            context[subtask.name] = result.output

            # Confidence-driven re-decomposition.
            conf = _extract_confidence(result.output)
            if (
                self._metacognition is not None
                and conf is not None
                and conf < self._metacognition.redecompose_threshold
                and redecomposition_count
                < self._metacognition.max_redecompositions
            ):
                # Surface the cognitive moment — when configured with
                # an event bus, emit a typed event so consumers
                # (Polymath's metacognition badge / healing log) can
                # render "the orchestrator just re-planned because
                # confidence dropped to X". Best-effort; bus failures
                # are logged but never propagate.
                if self._bus is not None:
                    try:
                        await self._bus.emit(
                            AgentEvent(
                                event_type="metacognition.redecomposition_triggered",
                                agent_name=self._planner.agent_name,
                                timestamp=datetime.now(UTC),
                                data={
                                    "subtask_name": subtask.name,
                                    "confidence": conf,
                                    "threshold": self._metacognition.redecompose_threshold,
                                    "attempt": redecomposition_count + 1,
                                    "max_attempts": self._metacognition.max_redecompositions,
                                    "remaining_subtasks": [
                                        s.name for s in subtasks[i + 1 :]
                                    ],
                                },
                            )
                        )
                    except Exception as exc:  # noqa: BLE001
                        logger.warning(
                            "metacognition redecomposition emit failed: {e}",
                            e=exc,
                        )

                new_remaining = await self._redecompose(
                    original_goal=goal,
                    completed=results,
                    remaining=subtasks[i + 1 :],
                    triggering_result=result,
                    triggering_confidence=conf,
                )
                subtasks = subtasks[: i + 1] + list(new_remaining)
                redecomposition_count += 1
        i += 1

    total_ms = (time.monotonic() - start) * 1000
    all_success = all(r.success for r in results)

    summary_parts = []
    for r in results:
        status = "OK" if r.success else "FAILED"
        summary_parts.append(
            f"  [{status}] {r.subtask_name} (via {r.agent_used})"
        )
    summary = f"Goal: {goal}\n" + "\n".join(summary_parts)

    return ExecutionResult(
        goal=goal,
        success=all_success,
        subtask_results=results,
        summary=summary,
        total_duration_ms=total_ms,
    )

SubTask

A single subtask derived from goal decomposition.

SubTaskResult

Result of a single subtask execution.

was_spawned instance-attribute

was_spawned: bool

True when the agent was newly created this turn (planner emitted a fresh AgentSpec and the factory hydrated it). False when the agent was already in the per-run cache or rehydrated from a memory hit — telemetry/UI use this to distinguish "we built something new" from "we reused what we had".

TaskDecomposition

Output of the goal decomposition step.

ToolInfo dataclass

ToolInfo(
    name: str,
    description: str,
    parameters: dict[str, Any] = dict(),
)

Metadata about a registered tool.

ToolRegistry

ToolRegistry()

Central registry of available tools.

Agents and the MetaOrchestrator discover tools here. Tools can be pre-registered by developers or added dynamically.

Initialize an empty registry.

Source code in orqest/autonomy/registry.py

def __init__(self) -> None:
    """Initialize an empty registry."""
    self._tools: dict[str, Tool] = {}
    self._info: dict[str, ToolInfo] = {}

register

register(
    tool: Tool, *, description: str | None = None
) -> None

Register a tool. Uses tool.name as the key.

Source code in orqest/autonomy/registry.py

def register(self, tool: Tool, *, description: str | None = None) -> None:
    """Register a tool. Uses tool.name as the key."""
    name = tool.name
    self._tools[name] = tool
    self._info[name] = ToolInfo(
        name=name,
        description=description or getattr(tool, "description", ""),
    )

get

get(name: str) -> Tool | None

Get a tool by name. Returns None if not found.

Source code in orqest/autonomy/registry.py

def get(self, name: str) -> Tool | None:
    """Get a tool by name. Returns None if not found."""
    return self._tools.get(name)

get_or_discover async

get_or_discover(
    name: str,
    *,
    discovery: MCPDiscovery | None = None,
    manager: MCPServerManager | None = None,
    permission: PermissionGate | None = None,
    audit_bus: EventBus | None = None,
    max_servers: int = 3,
) -> Tool | None

Get a tool by name; if missing, fall back to MCP discovery.

On a registry miss, searches for an MCP server advertising the tool, gates the request through permission (default :class:DenyAll), connects via manager, and registers the discovered tools transparently. Returns the matching tool if found, otherwise None.

Audit-log events emitted via audit_bus (when supplied):

• discovery.requested — a missing tool triggered discovery.
• discovery.denied — :class:PermissionGate rejected the request.
• discovery.connected — a discovered tool was registered (one per tool).
• discovery.failed — search/connect raised.

Parameters:

Name	Type	Description	Default
name	str	Requested tool name.	required
discovery	MCPDiscovery | None	Optional :class:MCPDiscovery. If absent, returns None.	None
manager	MCPServerManager | None	Optional :class:MCPServerManager for connecting. If absent, returns None.	None
permission	PermissionGate | None	Gate to require explicit approval. Default :class:DenyAll — discovery requires opt-in.	None
audit_bus	EventBus | None	Optional event bus for the audit trail.	None
max_servers	int	Maximum number of discovered servers to try before giving up.	3

Source code in orqest/autonomy/registry.py

async def get_or_discover(
    self,
    name: str,
    *,
    discovery: MCPDiscovery | None = None,
    manager: MCPServerManager | None = None,
    permission: PermissionGate | None = None,
    audit_bus: EventBus | None = None,
    max_servers: int = 3,
) -> Tool | None:
    """Get a tool by name; if missing, fall back to MCP discovery.

    On a registry miss, searches for an MCP server advertising the
    tool, gates the request through ``permission`` (default
    :class:`DenyAll`), connects via ``manager``, and registers the
    discovered tools transparently. Returns the matching tool if
    found, otherwise ``None``.

    Audit-log events emitted via ``audit_bus`` (when supplied):

    * ``discovery.requested`` — a missing tool triggered discovery.
    * ``discovery.denied`` — :class:`PermissionGate` rejected the request.
    * ``discovery.connected`` — a discovered tool was registered (one per tool).
    * ``discovery.failed`` — search/connect raised.

    Args:
        name: Requested tool name.
        discovery: Optional :class:`MCPDiscovery`. If absent, returns ``None``.
        manager: Optional :class:`MCPServerManager` for connecting.
            If absent, returns ``None``.
        permission: Gate to require explicit approval. Default
            :class:`DenyAll` — discovery requires opt-in.
        audit_bus: Optional event bus for the audit trail.
        max_servers: Maximum number of discovered servers to try
            before giving up.
    """
    tool = self._tools.get(name)
    if tool is not None:
        return tool
    if discovery is None or manager is None:
        return None

    from orqest.mcp.permission import DenyAll

    gate: PermissionGate = permission or DenyAll()
    if not await gate.allow(name):
        await _audit(
            audit_bus,
            "discovery.denied",
            {"requested": name, "reason": "permission"},
        )
        return None

    await _audit(audit_bus, "discovery.requested", {"requested": name})
    try:
        servers = await discovery.search(name, max_results=max_servers)
    except Exception as exc:
        await _audit(
            audit_bus,
            "discovery.failed",
            {"requested": name, "stage": "search", "error": str(exc)[:200]},
        )
        return None

    for server in servers:
        try:
            conn = await manager.connect(server.to_config())
        except Exception as exc:
            await _audit(
                audit_bus,
                "discovery.failed",
                {
                    "requested": name,
                    "server": server.name,
                    "stage": "connect",
                    "error": str(exc)[:200],
                },
            )
            continue
        for t in conn.tools:
            self.register(t, description=getattr(t, "description", ""))
            await _audit(
                audit_bus,
                "discovery.connected",
                {
                    "server": server.name,
                    "tool": t.name,
                    "source": getattr(server, "source", "unknown"),
                },
            )
        if name in self._tools:
            return self._tools[name]
    return None

search

search(query: str, *, k: int = 5) -> list[ToolInfo]

Search tools by keyword matching in name and description.

Source code in orqest/autonomy/registry.py

def search(self, query: str, *, k: int = 5) -> list[ToolInfo]:
    """Search tools by keyword matching in name and description."""
    query_lower = query.lower()
    matches: list[tuple[int, ToolInfo]] = []
    for info in self._info.values():
        score = 0
        if query_lower in info.name.lower():
            score += 2
        if query_lower in info.description.lower():
            score += 1
        if score > 0:
            matches.append((score, info))
    matches.sort(key=lambda x: x[0], reverse=True)
    return [info for _, info in matches[:k]]

list_all

list_all() -> list[ToolInfo]

List all registered tools.

Source code in orqest/autonomy/registry.py

def list_all(self) -> list[ToolInfo]:
    """List all registered tools."""
    return list(self._info.values())

remove

remove(name: str) -> None

Remove a tool by name. No error if not found.

Source code in orqest/autonomy/registry.py

def remove(self, name: str) -> None:
    """Remove a tool by name. No error if not found."""
    self._tools.pop(name, None)
    self._info.pop(name, None)

__len__

__len__() -> int

Return the number of registered tools.

Source code in orqest/autonomy/registry.py

def __len__(self) -> int:
    """Return the number of registered tools."""
    return len(self._tools)

__contains__

__contains__(name: str) -> bool

Check whether a tool with the given name is registered.

Source code in orqest/autonomy/registry.py

def __contains__(self, name: str) -> bool:
    """Check whether a tool with the given name is registered."""
    return name in self._tools

AgentSpec

Everything needed to spawn an agent at runtime.

An LLM produces this as structured output. The AgentFactory hydrates it into a live BaseAgent.

Output shape is declared via exactly one of:

• output_schema: a JSON Schema dict (the wire-format option — LLMs can emit this, and it survives serialization/persistence).
• output_type: a Pydantic BaseModel subclass (the code-side ergonomic option — terser than authoring JSON Schema by hand; not serializable, so use this for in-process construction).

A validator enforces "exactly one of" — neither raises, both raises.

output_schema class-attribute instance-attribute

output_schema: dict[str, Any] | None = None

JSON Schema dict describing the agent's structured output. Use this when the spec is emitted by an LLM or persisted across processes. Pair with the properties/required shape that :meth:AgentFactory._schema_to_model expects.

output_type class-attribute instance-attribute

output_type: type[BaseModel] | None = None

A Pydantic BaseModel subclass. Use this when constructing the spec in code — it's terser than authoring JSON Schema by hand and you get static-typing of the output. Not serializable; the JSON Schema path is the wire-format option.

tools class-attribute instance-attribute

tools: list[ToolSpec | GeneratedToolSpec] = Field(
    default_factory=list
)

Mixed list — pre-registered tool references (:class:ToolSpec) and runtime-generated tools (:class:GeneratedToolSpec). Pydantic v2 smart-union dispatches by structure: the implementation field on :class:GeneratedToolSpec is the disambiguator.

GeneratedToolSpec

A tool the LLM defines AT RUNTIME — implementation included.

The companion to :class:ToolSpec for cases where the agent needs a capability that does NOT yet exist in :class:ToolRegistry. The implementation string is the body of the tool function:

• It receives a single args dict (matching parameters)
• It must return a JSON-serializable value
• It executes inside the configured :class:orqest.sandbox.Sandbox

Hydration is performed by :class:orqest.autonomy.DynamicToolFactory, which validates the implementation (AST check + allowed-imports allowlist) before producing the runnable pydantic_ai.Tool.

The presence of implementation is the structural discriminator that distinguishes :class:GeneratedToolSpec from :class:ToolSpec in the :attr:AgentSpec.tools smart-union — Pydantic v2 picks this variant when implementation is present in the input dict.

implementation instance-attribute

implementation: str

Python source — the body of a function that takes args (dict) and returns a JSON-serializable value. Use return for the result.

Example::

"import re\n"
"matches = re.findall(r'\\d+', args['text'])\n"
"return {'matches': matches}\n"

allowed_imports class-attribute instance-attribute

allowed_imports: set[str] = Field(default_factory=set)

Top-level module names the implementation may import. Empty set rejects any import statement at validate time. Use sparingly — every entry is a safety surface.

dependencies class-attribute instance-attribute

dependencies: list[str] = Field(default_factory=list)

Optional pip specifiers (e.g. ["pandas>=2.0", "httpx"]) required by the implementation. Tier-2 :class:DockerSandbox installs them into the agent's per-agent .venv on first invocation, gated by the sandbox's allowed_packages allowlist (default-deny). Tier-0 / Tier-1 sandboxes ignore — they have no per-agent venv concept. Empty default keeps the field backward-compatible.

timeout_s class-attribute instance-attribute

timeout_s: float = Field(default=5.0, gt=0.0)

Per-invocation wall-clock cap inside the sandbox.

memory_mb class-attribute instance-attribute

memory_mb: int = Field(default=128, gt=0)

Per-invocation memory cap (RLIMIT_AS on POSIX). Ignored on Windows.

ToolSpec

Description of a tool an agent needs.

The name is resolved against :class:ToolRegistry at spawn time. parameters is a JSON-Schema-shaped dict carried for the LLM's benefit (it never reaches the registered tool — that contract lives on the tool itself).

DynamicToolFactory

DynamicToolFactory(
    sandbox: Sandbox,
    *,
    bus: EventBus | None = None,
    default_timeout_s: float = 5.0,
    default_memory_mb: int = 128,
)

Spawn :class:pydantic_ai.Tool objects from :class:GeneratedToolSpec.

Parameters:

Name	Type	Description	Default
sandbox	Sandbox	Required :class:Sandbox backend (typically :class:SubprocessSandbox for production, :class:InProcessSandbox(unsafe=True) for tests).	required
bus	EventBus | None	Optional :class:EventBus for tool.* / sandbox.validation_rejected events.	None
default_timeout_s	float	Fallback timeout when a spec doesn't override.	5.0
default_memory_mb	int	Fallback memory cap when a spec doesn't override.	128

Source code in orqest/autonomy/tool_factory.py

def __init__(
    self,
    sandbox: Sandbox,
    *,
    bus: EventBus | None = None,
    default_timeout_s: float = 5.0,
    default_memory_mb: int = 128,
) -> None:
    self._sandbox = sandbox
    self._bus = bus
    self._default_timeout_s = default_timeout_s
    self._default_memory_mb = default_memory_mb

sandbox property

sandbox: Sandbox

The sandbox instance bound at construction time.

spawn async

spawn(
    spec: GeneratedToolSpec, *, agent_id: str | None = None
) -> Tool

Validate the spec, return a runnable :class:pydantic_ai.Tool.

Parameters:

Name	Type	Description	Default
spec	GeneratedToolSpec	The :class:GeneratedToolSpec carrying the implementation + parameters + safety knobs.	required
agent_id	str | None	Optional agent identifier — Tier-2 :class:DockerSandbox routes execution into the agent's per-agent subfolder + .venv. Tier-0 / Tier-1 ignore. Captured into the runner closure so every invocation of the returned Tool carries the same agent_id.	None

Raises:

Type	Description
ValidationError	When the implementation fails static checks (disallowed import, forbidden built-in, syntax error).

Source code in orqest/autonomy/tool_factory.py

async def spawn(
    self,
    spec: GeneratedToolSpec,
    *,
    agent_id: str | None = None,
) -> Tool:
    """Validate the spec, return a runnable :class:`pydantic_ai.Tool`.

    Args:
        spec: The :class:`GeneratedToolSpec` carrying the implementation
            + parameters + safety knobs.
        agent_id: Optional agent identifier — Tier-2
            :class:`DockerSandbox` routes execution into the agent's
            per-agent subfolder + ``.venv``. Tier-0 / Tier-1 ignore.
            Captured into the runner closure so every invocation of the
            returned ``Tool`` carries the same ``agent_id``.

    Raises:
        ValidationError: When the implementation fails static checks
            (disallowed import, forbidden built-in, syntax error).

    """
    try:
        await self._sandbox.validate(
            spec.implementation, allowed_imports=spec.allowed_imports
        )
    except ValidationError as exc:
        self._emit(
            "sandbox.validation_rejected",
            tool_name=spec.name,
            reason=str(exc),
        )
        self._emit(
            "tool.spawn_failed",
            tool_name=spec.name,
            reason=str(exc),
        )
        raise

    # Bind the spec into the closure so the runner has everything it
    # needs at invocation time.
    timeout_s = spec.timeout_s if spec.timeout_s else self._default_timeout_s
    memory_mb = spec.memory_mb if spec.memory_mb else self._default_memory_mb
    dependencies = list(spec.dependencies) if spec.dependencies else None
    bound_agent_id = agent_id
    sandbox = self._sandbox
    emit = self._emit

    async def _runner(**kwargs: Any) -> Any:
        """Spawned tool body — see DynamicToolFactory.spawn docstring.

        Pydantic-ai's tool loop passes LLM-supplied kwargs here; we
        forward them as the ``args`` dict the sandbox expects.
        """
        result = await sandbox.execute(
            spec.implementation,
            args=dict(kwargs),
            allowed_imports=spec.allowed_imports,
            timeout_s=timeout_s,
            memory_mb=memory_mb,
            agent_id=bound_agent_id,
            dependencies=dependencies,
        )
        if result.success:
            emit(
                "tool.invocation_completed",
                tool_name=spec.name,
                duration_ms=result.duration_ms,
            )
            return result.output
        emit(
            "tool.invocation_failed",
            tool_name=spec.name,
            error=(result.error or "")[:300],
            duration_ms=result.duration_ms,
        )
        # Return a structured error so the LLM agent loop sees it as a
        # tool result (not a Python exception that would crash the loop).
        return {
            "error": result.error or "sandbox execution failed",
            "stage": "sandbox.execute",
            "tool_name": spec.name,
        }

    # Set name + docstring so pydantic-ai surfaces them sensibly.
    _runner.__name__ = spec.name
    _runner.__doc__ = spec.description

    tool = Tool(_runner, name=spec.name, description=spec.description)
    self._emit("tool.spawned", tool_name=spec.name)
    return tool