SDK

The SDK module provides the high-level Python interface for OpenJarvis. The Jarvis class wraps engine discovery, agent dispatch, memory operations, and telemetry into a single convenient entry point. MemoryHandle acts as a lazy proxy for memory backend operations.

Jarvis

Jarvis

Jarvis(*, config: Optional[JarvisConfig] = None, config_path: Optional[str] = None, engine_key: Optional[str] = None, model: Optional[str] = None)

High-level OpenJarvis SDK.

Usage::

from openjarvis import Jarvis

j = Jarvis()
response = j.ask("Hello, what can you do?")
print(response)
j.close()

Source code in src/openjarvis/sdk.py

def __init__(
    self,
    *,
    config: Optional[JarvisConfig] = None,
    config_path: Optional[str] = None,
    engine_key: Optional[str] = None,
    model: Optional[str] = None,
) -> None:
    if config is not None:
        self._config = config
    elif config_path is not None:
        self._config = load_config(Path(config_path))
    else:
        self._config = load_config()

    self._engine_key = engine_key
    self._model_override = model
    self._engine: Any = None
    self._resolved_engine_key: Optional[str] = None
    self._bus = EventBus()
    self._telem_store: Optional[TelemetryStore] = None
    self._audit_logger: Any = None
    self.memory = MemoryHandle(self._config)

    # Set up telemetry
    if self._config.telemetry.enabled:
        try:
            self._telem_store = TelemetryStore(self._config.telemetry.db_path)
            self._telem_store.subscribe_to_bus(self._bus)
        except Exception:
            pass

    # Set up security audit logger
    if self._config.security.enabled:
        try:
            from openjarvis.security.audit import AuditLogger

            self._audit_logger = AuditLogger(
                db_path=self._config.security.audit_log_path,
                bus=self._bus,
            )
        except Exception:
            pass

Attributes

config property

config: JarvisConfig

Return the active configuration.

version property

version: str

Return the OpenJarvis version string.

Functions

ask

ask(query: str, *, model: Optional[str] = None, agent: Optional[str] = None, tools: Optional[List[str]] = None, temperature: Optional[float] = None, max_tokens: Optional[int] = None, context: bool = True) -> str

Send a query and return the response text.

Source code in src/openjarvis/sdk.py

def ask(
    self,
    query: str,
    *,
    model: Optional[str] = None,
    agent: Optional[str] = None,
    tools: Optional[List[str]] = None,
    temperature: Optional[float] = None,
    max_tokens: Optional[int] = None,
    context: bool = True,
) -> str:
    """Send a query and return the response text."""
    result = self.ask_full(
        query,
        model=model,
        agent=agent,
        tools=tools,
        temperature=temperature,
        max_tokens=max_tokens,
        context=context,
    )
    return result["content"]

ask_full

ask_full(query: str, *, model: Optional[str] = None, agent: Optional[str] = None, tools: Optional[List[str]] = None, temperature: Optional[float] = None, max_tokens: Optional[int] = None, context: bool = True) -> Dict[str, Any]

Send a query and return the full result dict.

Returns a dict with keys: content, usage, tool_results (if agent mode).

Source code in src/openjarvis/sdk.py

def ask_full(
    self,
    query: str,
    *,
    model: Optional[str] = None,
    agent: Optional[str] = None,
    tools: Optional[List[str]] = None,
    temperature: Optional[float] = None,
    max_tokens: Optional[int] = None,
    context: bool = True,
) -> Dict[str, Any]:
    """Send a query and return the full result dict.

    Returns a dict with keys: content, usage, tool_results (if agent mode).
    """
    self._ensure_engine()
    if temperature is None:
        temperature = self._config.intelligence.temperature
    if max_tokens is None:
        max_tokens = self._config.intelligence.max_tokens

    model_name = model or self._model_override

    # Resolve model via router if not specified
    if model_name is None:
        model_name = self._resolve_model(query)

    if not model_name:
        models = self._engine.list_models()
        model_name = models[0] if models else "default"

    # Agent mode
    if agent is not None:
        return self._run_agent(
            agent, query, model_name,
            tools=tools or [],
            temperature=temperature,
            max_tokens=max_tokens,
            context=context,
        )

    # Direct engine mode
    messages = [Message(role=Role.USER, content=query)]

    # Memory context injection
    if context and self._config.agent.context_from_memory:
        messages = self._inject_context(query, messages)

    result = instrumented_generate(
        self._engine,
        messages,
        model=model_name,
        bus=self._bus,
        temperature=temperature,
        max_tokens=max_tokens,
    )

    return {
        "content": result.get("content", ""),
        "usage": result.get("usage", {}),
        "model": model_name,
        "engine": self._resolved_engine_key,
    }

list_models

list_models() -> List[str]

Return a list of available model identifiers.

Source code in src/openjarvis/sdk.py

def list_models(self) -> List[str]:
    """Return a list of available model identifiers."""
    self._ensure_engine()
    return self._engine.list_models()

list_engines

list_engines() -> List[str]

Return a list of registered engine keys.

Source code in src/openjarvis/sdk.py

def list_engines(self) -> List[str]:
    """Return a list of registered engine keys."""
    from openjarvis.core.registry import EngineRegistry

    return list(EngineRegistry.keys())

close

close() -> None

Release all resources.

Source code in src/openjarvis/sdk.py

def close(self) -> None:
    """Release all resources."""
    self.memory.close()
    if self._telem_store is not None:
        try:
            self._telem_store.close()
        except Exception:
            pass
        self._telem_store = None
    if self._audit_logger is not None:
        try:
            self._audit_logger.close()
        except Exception:
            pass
        self._audit_logger = None
    self._engine = None

MemoryHandle

MemoryHandle

MemoryHandle(config: JarvisConfig)

Proxy for memory operations. Lazily initializes backend.

Source code in src/openjarvis/sdk.py

def __init__(self, config: JarvisConfig) -> None:
    self._config = config
    self._backend: Any = None

Functions

index

index(path: str, *, chunk_size: int = 512, chunk_overlap: int = 64) -> Dict[str, Any]

Index a file or directory into memory.

Source code in src/openjarvis/sdk.py

def index(
    self,
    path: str,
    *,
    chunk_size: int = 512,
    chunk_overlap: int = 64,
) -> Dict[str, Any]:
    """Index a file or directory into memory."""
    from openjarvis.memory.chunking import ChunkConfig
    from openjarvis.memory.ingest import ingest_path

    backend = self._get_backend()
    cfg = ChunkConfig(chunk_size=chunk_size, chunk_overlap=chunk_overlap)
    chunks = ingest_path(Path(path), config=cfg)

    doc_ids: List[str] = []
    for chunk in chunks:
        doc_id = backend.store(
            chunk.content, source=chunk.source,
            metadata={"index": chunk.index},
        )
        doc_ids.append(doc_id)

    return {
        "chunks": len(chunks),
        "doc_ids": doc_ids,
        "path": path,
    }

search

search(query: str, *, top_k: int = 5) -> List[Dict[str, Any]]

Search memory for relevant chunks.

Source code in src/openjarvis/sdk.py

def search(self, query: str, *, top_k: int = 5) -> List[Dict[str, Any]]:
    """Search memory for relevant chunks."""
    backend = self._get_backend()
    results = backend.retrieve(query, top_k=top_k)
    return [
        {
            "content": r.content,
            "score": r.score,
            "source": r.source,
            "metadata": r.metadata,
        }
        for r in results
    ]

stats

stats() -> Dict[str, Any]

Return memory backend statistics.

Source code in src/openjarvis/sdk.py

def stats(self) -> Dict[str, Any]:
    """Return memory backend statistics."""
    backend = self._get_backend()
    if hasattr(backend, "count"):
        return {
            "count": backend.count(),
            "backend": self._config.memory.default_backend,
        }
    return {"backend": self._config.memory.default_backend}

close

close() -> None

Release the memory backend.

Source code in src/openjarvis/sdk.py

def close(self) -> None:
    """Release the memory backend."""
    if self._backend is not None:
        if hasattr(self._backend, "close"):
            self._backend.close()
        self._backend = None