Index

agents

Agents primitive — multi-turn reasoning and tool use.

Classes

AgentContext dataclass

AgentContext(conversation: Conversation = Conversation(), tools: List[str] = list(), memory_results: List[Any] = list(), metadata: Dict[str, Any] = dict())

Runtime context handed to an agent on each invocation.

AgentResult dataclass

AgentResult(content: str, tool_results: List[ToolResult] = list(), turns: int = 0, metadata: Dict[str, Any] = dict())

Result returned after an agent completes a run.

BaseAgent

BaseAgent(engine: InferenceEngine, model: str, *, bus: Optional[EventBus] = None, temperature: Optional[float] = None, max_tokens: Optional[int] = None, prompt_builder: Optional[Any] = None)

Bases: ABC

Base class for all agent implementations.

Subclasses must be registered via @AgentRegistry.register("name") to become discoverable.

Provides concrete helper methods that eliminate boilerplate in subclasses:

• :meth:_emit_turn_start / :meth:_emit_turn_end -- event bus
• :meth:_build_messages -- conversation + system prompt assembly
• :meth:_generate -- delegates to engine with stored defaults
• :meth:_max_turns_result -- standard max-turns-exceeded result
• :meth:_strip_think_tags -- remove <think> blocks

Source code in src/openjarvis/agents/_stubs.py

def __init__(
    self,
    engine: InferenceEngine,
    model: str,
    *,
    bus: Optional[EventBus] = None,
    temperature: Optional[float] = None,
    max_tokens: Optional[int] = None,
    prompt_builder: Optional[Any] = None,
) -> None:
    self._engine = engine
    self._model = model
    self._bus = bus
    self._prompt_builder = prompt_builder

    # Three-tier resolution: explicit arg > config > class default > hardcoded
    if temperature is not None and max_tokens is not None:
        self._temperature = temperature
        self._max_tokens = max_tokens
    else:
        try:
            cfg = load_config()
            self._temperature = (
                temperature
                if temperature is not None
                else cfg.intelligence.temperature
            )
            self._max_tokens = (
                max_tokens
                if max_tokens is not None
                else cfg.intelligence.max_tokens
            )
        except Exception:
            self._temperature = (
                temperature
                if temperature is not None
                else getattr(self, "_default_temperature", 0.7)
            )
            self._max_tokens = (
                max_tokens
                if max_tokens is not None
                else getattr(self, "_default_max_tokens", 1024)
            )

Functions

run abstractmethod

run(input: str, context: Optional[AgentContext] = None, **kwargs: Any) -> AgentResult

Execute the agent on input and return an AgentResult.

Source code in src/openjarvis/agents/_stubs.py

@abstractmethod
def run(
    self,
    input: str,
    context: Optional[AgentContext] = None,
    **kwargs: Any,
) -> AgentResult:
    """Execute the agent on *input* and return an ``AgentResult``."""

ToolUsingAgent

ToolUsingAgent(engine: InferenceEngine, model: str, *, tools: Optional[List['BaseTool']] = None, bus: Optional[EventBus] = None, max_turns: Optional[int] = None, temperature: Optional[float] = None, max_tokens: Optional[int] = None, loop_guard_config: Optional[Any] = None, capability_policy: Optional[Any] = None, agent_id: Optional[str] = None, interactive: bool = False, confirm_callback: Optional[Any] = None)

Bases: BaseAgent

Intermediate base for agents that accept and use tools.

Sets accepts_tools = True for CLI/SDK introspection, and initialises a :class:ToolExecutor from the provided tools.

Source code in src/openjarvis/agents/_stubs.py

def __init__(
    self,
    engine: InferenceEngine,
    model: str,
    *,
    tools: Optional[List["BaseTool"]] = None,  # noqa: F821
    bus: Optional[EventBus] = None,
    max_turns: Optional[int] = None,
    temperature: Optional[float] = None,
    max_tokens: Optional[int] = None,
    loop_guard_config: Optional[Any] = None,
    capability_policy: Optional[Any] = None,
    agent_id: Optional[str] = None,
    interactive: bool = False,
    confirm_callback: Optional[Any] = None,
) -> None:
    super().__init__(
        engine,
        model,
        bus=bus,
        temperature=temperature,
        max_tokens=max_tokens,
    )
    from openjarvis.tools._stubs import ToolExecutor

    self._tools = tools or []
    _aid = agent_id or getattr(self, "agent_id", "")
    self._executor = ToolExecutor(
        self._tools,
        bus=bus,
        capability_policy=capability_policy,
        agent_id=_aid,
        interactive=interactive,
        confirm_callback=confirm_callback,
    )
    # Resolve max_turns: explicit arg > config > class default > 10
    if max_turns is not None:
        self._max_turns = max_turns
    else:
        try:
            cfg = load_config()
            self._max_turns = cfg.agent.max_turns
        except Exception:
            self._max_turns = getattr(self, "_default_max_turns", 10)

    # Loop guard
    self._loop_guard = None
    try:
        from openjarvis.agents.loop_guard import LoopGuard, LoopGuardConfig

        if loop_guard_config is None:
            loop_guard_config = LoopGuardConfig()
        elif isinstance(loop_guard_config, dict):
            loop_guard_config = LoopGuardConfig(**loop_guard_config)
        if loop_guard_config.enabled:
            self._loop_guard = LoopGuard(loop_guard_config, bus=bus)
    except ImportError:
        pass