Index

operators

Operators — persistent, scheduled autonomous agents.

Classes

OperatorManager

OperatorManager(system: Any)

Manages operator manifests and their lifecycle via the TaskScheduler.

PARAMETER	DESCRIPTION
system	A JarvisSystem instance (used to access scheduler, session_store, memory_backend, and to run operators via system.ask()). TYPE: Any

Source code in src/openjarvis/operators/manager.py

def __init__(self, system: Any) -> None:
    self._system = system
    self._manifests: Dict[str, OperatorManifest] = {}

Attributes

manifests property

manifests: Dict[str, OperatorManifest]

All registered manifests.

Functions

register

register(manifest: OperatorManifest) -> None

Register an operator manifest.

Source code in src/openjarvis/operators/manager.py

def register(self, manifest: OperatorManifest) -> None:
    """Register an operator manifest."""
    self._manifests[manifest.id] = manifest
    logger.info("Registered operator: %s", manifest.id)

discover

discover(directory: str | Path) -> List[OperatorManifest]

Discover and register operator manifests from a directory.

Scans for *.toml files in directory and loads each as an operator manifest.

Source code in src/openjarvis/operators/manager.py

def discover(self, directory: str | Path) -> List[OperatorManifest]:
    """Discover and register operator manifests from a directory.

    Scans for ``*.toml`` files in *directory* and loads each as an
    operator manifest.
    """
    directory = Path(directory).expanduser()
    found: List[OperatorManifest] = []
    if not directory.is_dir():
        return found
    for toml_path in sorted(directory.glob("*.toml")):
        try:
            manifest = load_operator(toml_path)
            self.register(manifest)
            found.append(manifest)
        except Exception:
            logger.warning("Failed to load operator from %s", toml_path)
    return found

activate

activate(operator_id: str) -> str

Activate an operator by creating a scheduler task.

Returns the scheduler task ID (deterministic: operator:{id}).

Raises KeyError if the operator is not registered, or RuntimeError if the scheduler is not available.

Source code in src/openjarvis/operators/manager.py

def activate(self, operator_id: str) -> str:
    """Activate an operator by creating a scheduler task.

    Returns the scheduler task ID (deterministic: ``operator:{id}``).

    Raises ``KeyError`` if the operator is not registered, or
    ``RuntimeError`` if the scheduler is not available.
    """
    manifest = self._manifests.get(operator_id)
    if manifest is None:
        raise KeyError(f"Operator not registered: {operator_id}")

    scheduler = self._system.scheduler
    if scheduler is None:
        raise RuntimeError(
            "TaskScheduler not available. Enable [scheduler] in config."
        )

    task_id = f"operator:{operator_id}"

    # Check if already active
    try:
        existing = scheduler.list_tasks()
        for t in existing:
            if t.id == task_id and t.status == "active":
                logger.info("Operator %s already active", operator_id)
                return task_id
    except Exception:
        pass

    tools_str = ",".join(manifest.tools) if manifest.tools else ""

    metadata: Dict[str, Any] = {
        "operator_id": operator_id,
        "system_prompt": manifest.system_prompt,
        "temperature": manifest.temperature,
        "max_turns": manifest.max_turns,
    }

    # Use the scheduler's create_task but with a deterministic ID
    task = scheduler.create_task(
        prompt=_TICK_PROMPT,
        schedule_type=manifest.schedule_type,
        schedule_value=manifest.schedule_value,
        agent="operative",
        tools=tools_str,
        metadata=metadata,
    )
    # Override the random ID with our deterministic one
    task_dict = task.to_dict()
    task_dict["id"] = task_id
    scheduler._store.save_task(task_dict)

    logger.info("Activated operator %s (task_id=%s)", operator_id, task_id)
    return task_id

deactivate

deactivate(operator_id: str) -> None

Deactivate an operator by cancelling its scheduler task.

Source code in src/openjarvis/operators/manager.py

def deactivate(self, operator_id: str) -> None:
    """Deactivate an operator by cancelling its scheduler task."""
    scheduler = self._system.scheduler
    if scheduler is None:
        raise RuntimeError("TaskScheduler not available.")
    task_id = f"operator:{operator_id}"
    try:
        scheduler.cancel_task(task_id)
        logger.info("Deactivated operator %s", operator_id)
    except KeyError:
        logger.warning("No active task for operator %s", operator_id)

pause

pause(operator_id: str) -> None

Pause an active operator.

Source code in src/openjarvis/operators/manager.py

def pause(self, operator_id: str) -> None:
    """Pause an active operator."""
    scheduler = self._system.scheduler
    if scheduler is None:
        raise RuntimeError("TaskScheduler not available.")
    scheduler.pause_task(f"operator:{operator_id}")
    logger.info("Paused operator %s", operator_id)

resume

resume(operator_id: str) -> None

Resume a paused operator.

Source code in src/openjarvis/operators/manager.py

def resume(self, operator_id: str) -> None:
    """Resume a paused operator."""
    scheduler = self._system.scheduler
    if scheduler is None:
        raise RuntimeError("TaskScheduler not available.")
    scheduler.resume_task(f"operator:{operator_id}")
    logger.info("Resumed operator %s", operator_id)

status

status() -> List[Dict[str, Any]]

Return status of all registered operators.

Merges manifest info with scheduler task state.

Source code in src/openjarvis/operators/manager.py

def status(self) -> List[Dict[str, Any]]:
    """Return status of all registered operators.

    Merges manifest info with scheduler task state.
    """
    results: List[Dict[str, Any]] = []
    scheduler = self._system.scheduler

    for op_id, manifest in self._manifests.items():
        info: Dict[str, Any] = {
            "id": op_id,
            "name": manifest.name,
            "description": manifest.description,
            "schedule_type": manifest.schedule_type,
            "schedule_value": manifest.schedule_value,
            "tools": manifest.tools,
            "status": "registered",
            "next_run": None,
            "last_run": None,
        }

        if scheduler is not None:
            task_id = f"operator:{op_id}"
            try:
                tasks = scheduler.list_tasks()
                for t in tasks:
                    if t.id == task_id:
                        info["status"] = t.status
                        info["next_run"] = t.next_run
                        info["last_run"] = t.last_run
                        break
            except Exception:
                pass

        results.append(info)
    return results

run_once

run_once(operator_id: str) -> str

Execute a single tick of an operator immediately.

Useful for development and testing. Returns the agent's response.

Source code in src/openjarvis/operators/manager.py

def run_once(self, operator_id: str) -> str:
    """Execute a single tick of an operator immediately.

    Useful for development and testing. Returns the agent's response.
    """
    manifest = self._manifests.get(operator_id)
    if manifest is None:
        raise KeyError(f"Operator not registered: {operator_id}")

    tools_list = manifest.tools if manifest.tools else None
    result = self._system.ask(
        _TICK_PROMPT,
        agent="operative",
        tools=tools_list,
        system_prompt=manifest.system_prompt,
        operator_id=operator_id,
        temperature=manifest.temperature,
    )
    if isinstance(result, dict):
        return result.get("content", str(result))
    return str(result)

get_manifest

get_manifest(operator_id: str) -> Optional[OperatorManifest]

Return the manifest for an operator, or None.

Source code in src/openjarvis/operators/manager.py

def get_manifest(self, operator_id: str) -> Optional[OperatorManifest]:
    """Return the manifest for an operator, or None."""
    return self._manifests.get(operator_id)

OperatorManifest dataclass

OperatorManifest(id: str, name: str, version: str = '0.1.0', description: str = '', author: str = '', tools: List[str] = list(), system_prompt: str = '', system_prompt_path: str = '', max_turns: int = 20, temperature: float = 0.3, schedule_type: str = 'interval', schedule_value: str = '300', metrics: List[str] = list(), required_capabilities: List[str] = list(), settings: Dict[str, Any] = dict(), metadata: Dict[str, Any] = dict())

Manifest describing a persistent autonomous operator.

Functions

load_operator

load_operator(path: str | Path) -> OperatorManifest

Load an operator manifest from a TOML file.

Supports inline system_prompt or external system_prompt_path (resolved relative to the TOML file).

Source code in src/openjarvis/operators/loader.py

def load_operator(path: str | Path) -> OperatorManifest:
    """Load an operator manifest from a TOML file.

    Supports inline ``system_prompt`` or external ``system_prompt_path``
    (resolved relative to the TOML file).
    """
    path = Path(path)
    with open(path, "rb") as fh:
        data = tomllib.load(fh)

    op_data = data.get("operator", {})

    # Resolve schedule sub-table
    schedule = op_data.get("schedule", {})
    schedule_type = schedule.get("type", op_data.get("schedule_type", "interval"))
    schedule_value = schedule.get("value", op_data.get("schedule_value", "300"))

    # Resolve agent sub-table
    agent_data = op_data.get("agent", {})
    tools = agent_data.get("tools", op_data.get("tools", []))
    max_turns = agent_data.get("max_turns", op_data.get("max_turns", 20))
    temperature = agent_data.get("temperature", op_data.get("temperature", 0.3))
    system_prompt = agent_data.get("system_prompt", op_data.get("system_prompt", ""))
    system_prompt_path = agent_data.get(
        "system_prompt_path", op_data.get("system_prompt_path", ""),
    )

    # Load external system prompt if specified and inline is empty
    if not system_prompt and system_prompt_path:
        prompt_path = Path(system_prompt_path)
        if not prompt_path.is_absolute():
            prompt_path = path.parent / prompt_path
        if prompt_path.exists():
            system_prompt = prompt_path.read_text(encoding="utf-8")

    return OperatorManifest(
        id=op_data.get("id", path.stem),
        name=op_data.get("name", path.stem),
        version=op_data.get("version", "0.1.0"),
        description=op_data.get("description", ""),
        author=op_data.get("author", ""),
        tools=tools,
        system_prompt=system_prompt,
        system_prompt_path=system_prompt_path,
        max_turns=max_turns,
        temperature=temperature,
        schedule_type=schedule_type,
        schedule_value=str(schedule_value),
        metrics=op_data.get("metrics", []),
        required_capabilities=op_data.get("required_capabilities", []),
        settings=op_data.get("settings", {}),
        metadata=op_data.get("metadata", {}),
    )