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: float = 0.7, max_tokens: int = 1024)

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: float = 0.7,
    max_tokens: int = 1024,
) -> None:
    self._engine = engine
    self._model = model
    self._bus = bus
    self._temperature = temperature
    self._max_tokens = max_tokens

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: int = 10, temperature: float = 0.7, max_tokens: int = 1024, loop_guard_config: Optional[Any] = None, capability_policy: Optional[Any] = None, agent_id: Optional[str] = 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: int = 10,
    temperature: float = 0.7,
    max_tokens: int = 1024,
    loop_guard_config: Optional[Any] = None,
    capability_policy: Optional[Any] = None,
    agent_id: Optional[str] = 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,
    )
    self._max_turns = max_turns

    # 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