orchestrata_ultimate.core.protocols.agent

Reference › orchestrata_ultimate › core › protocols › agent

VolkProtocol is the contract every volk (specialised AI agent) must satisfy to take part in the coordination system. It is a runtime_checkable Protocol with three read-only properties, volk_id (for example U:Cx), alias (the short routing form such as Cx) and role, and two coroutines: process_intent, which takes an intent dictionary and returns a result dictionary, and health_check, which returns a boolean. Agent runtimes, the tool registry and the volk registry import it through the protocols package, and interfaces.volk keeps the old IVolk name for it.

At a glance

Class diagram

Dependency diagram

API

Classes

VolkProtocol (protocol)

Bases: Protocol · Decorators: runtime_checkable · Line 9

Protocol interface for Volk agents.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/core/protocols/agent.py

"""Volk protocol — the contract every Volk agent must satisfy."""

from typing import Any, Protocol, runtime_checkable

from orchestrata_ultimate.core.base.traces import traces


@runtime_checkable
class VolkProtocol(Protocol):
    """Protocol interface for Volk agents.

    Every Volk (specialized AI agent) must implement this interface
    to participate in the Orchestrata coordination system.
    """

    @property
    def volk_id(self) -> str:
        """Unique identifier for this Volk (e.g., 'U:Cx')."""
        ...

    @property
    def alias(self) -> str:
        """Short alias for routing (e.g., 'Cx')."""
        ...

    @property
    def role(self) -> str:
        """Human-readable role description."""
        ...

    async def process_intent(self, intent: dict[str, Any]) -> dict[str, Any]:
        """Process an intent routed to this Volk.

        Args:
            intent: A dictionary containing at minimum 'action' and 'payload' keys.

        Returns:
            A dictionary with 'status' and 'result' keys.
        """
        ...

    @traces(spec="001", story="US-001", reqs=["REQ-001-018"])
    async def health_check(self) -> bool:
        """Return True if this Volk is operational."""
        ...