Server Module

The server module provides an OpenAI-compatible API server built on FastAPI and uvicorn. It exposes POST /v1/chat/completions (with streaming via SSE), GET /v1/models, and GET /health endpoints. Pydantic models ensure request/response validation matching the OpenAI API format.

Application Factory

create_app

create_app

create_app(engine, model: str, *, agent=None, bus=None, engine_name: str = '', agent_name: str = '', channel_bridge=None, config=None) -> FastAPI

Create and configure the FastAPI application.

PARAMETER	DESCRIPTION
engine	The inference engine to use for completions.
model	Default model name. TYPE: str
agent	Optional agent instance for agent-mode completions. DEFAULT: None
bus	Optional event bus for telemetry. DEFAULT: None
channel_bridge	Optional channel bridge for multi-platform messaging. DEFAULT: None
config	Optional JarvisConfig for security guardrails and other settings. DEFAULT: None

Source code in src/openjarvis/server/app.py

def create_app(
    engine,
    model: str,
    *,
    agent=None,
    bus=None,
    engine_name: str = "",
    agent_name: str = "",
    channel_bridge=None,
    config=None,
) -> FastAPI:
    """Create and configure the FastAPI application.

    Parameters
    ----------
    engine:
        The inference engine to use for completions.
    model:
        Default model name.
    agent:
        Optional agent instance for agent-mode completions.
    bus:
        Optional event bus for telemetry.
    channel_bridge:
        Optional channel bridge for multi-platform messaging.
    config:
        Optional JarvisConfig for security guardrails and other settings.
    """
    # Wrap engine with security guardrails if configured
    security_enabled = (
        config is not None
        and getattr(config, "security", None)
        and config.security.enabled
    )
    if security_enabled:
        try:
            from openjarvis.security.guardrails import GuardrailsEngine
            from openjarvis.security.scanner import PIIScanner, SecretScanner
            from openjarvis.security.types import RedactionMode

            scanners = []
            if config.security.secret_scanner:
                scanners.append(SecretScanner())
            if config.security.pii_scanner:
                scanners.append(PIIScanner())
            if scanners:
                mode = RedactionMode(config.security.mode)
                engine = GuardrailsEngine(
                    engine,
                    scanners=scanners,
                    mode=mode,
                    scan_input=config.security.scan_input,
                    scan_output=config.security.scan_output,
                    bus=bus,
                )
        except Exception:
            pass  # security is best-effort

    app = FastAPI(
        title="OpenJarvis API",
        description="OpenAI-compatible API server for OpenJarvis",
        version="1.0.0",
    )

    # Store dependencies in app state
    app.state.engine = engine
    app.state.model = model
    app.state.agent = agent
    app.state.bus = bus
    app.state.engine_name = engine_name
    app.state.agent_name = agent_name or (getattr(agent, "agent_id", None) if agent else None)
    app.state.channel_bridge = channel_bridge
    app.state.session_start = time.time()

    app.include_router(router)
    app.include_router(dashboard_router)

    # Serve static frontend assets if the static/ directory exists
    static_dir = pathlib.Path(__file__).parent / "static"
    if static_dir.is_dir():
        assets_dir = static_dir / "assets"
        if assets_dir.is_dir():
            app.mount(
                "/assets",
                _NoCacheStaticFiles(directory=assets_dir),
                name="static-assets",
            )

        @app.get("/{full_path:path}")
        async def spa_catch_all(full_path: str):
            """Serve index.html for SPA routes not handled by API endpoints."""
            return FileResponse(
                static_dir / "index.html",
                headers=_NO_CACHE_HEADERS,
            )

    return app

---

Route Handlers

chat_completions

chat_completions async

chat_completions(request_body: ChatCompletionRequest, request: Request)

Handle chat completion requests (streaming and non-streaming).

Source code in src/openjarvis/server/routes.py

@router.post("/v1/chat/completions")
async def chat_completions(request_body: ChatCompletionRequest, request: Request):
    """Handle chat completion requests (streaming and non-streaming)."""
    engine = request.app.state.engine
    agent = getattr(request.app.state, "agent", None)
    model = request_body.model

    if request_body.stream:
        bus = getattr(request.app.state, "bus", None)
        if agent is not None and bus is not None:
            return await _handle_agent_stream(agent, bus, model, request_body)
        return await _handle_stream(engine, model, request_body)

    # Non-streaming: use agent if available, otherwise direct engine call
    if agent is not None:
        return _handle_agent(agent, model, request_body)

    bus = getattr(request.app.state, "bus", None)
    return _handle_direct(engine, model, request_body, bus=bus)

list_models

list_models async

list_models(request: Request) -> ModelListResponse

List available models from the engine.

Source code in src/openjarvis/server/routes.py

@router.get("/v1/models")
async def list_models(request: Request) -> ModelListResponse:
    """List available models from the engine."""
    engine = request.app.state.engine
    model_ids = engine.list_models()
    return ModelListResponse(
        data=[ModelObject(id=mid) for mid in model_ids],
    )

health

health async

health(request: Request)

Health check endpoint.

Source code in src/openjarvis/server/routes.py

@router.get("/health")
async def health(request: Request):
    """Health check endpoint."""
    engine = request.app.state.engine
    healthy = engine.health()
    if not healthy:
        raise HTTPException(status_code=503, detail="Engine unhealthy")
    return {"status": "ok"}

---

Pydantic Request/Response Models

ChatMessage

ChatMessage

Bases: BaseModel

ChatCompletionRequest

ChatCompletionRequest

Bases: BaseModel

ChatCompletionResponse

ChatCompletionResponse

Bases: BaseModel

Choice

Choice

Bases: BaseModel

ChoiceMessage

ChoiceMessage

Bases: BaseModel

UsageInfo

UsageInfo

Bases: BaseModel

ChatCompletionChunk

ChatCompletionChunk

Bases: BaseModel

DeltaMessage

DeltaMessage

Bases: BaseModel

StreamChoice

StreamChoice

Bases: BaseModel

ModelObject

ModelObject

Bases: BaseModel

ModelListResponse

ModelListResponse

Bases: BaseModel

---

Channel Endpoints

list_channels

list_channels async

list_channels(request: Request)

List available messaging channels.

Source code in src/openjarvis/server/routes.py

@router.get("/v1/channels")
async def list_channels(request: Request):
    """List available messaging channels."""
    bridge = getattr(request.app.state, "channel_bridge", None)
    if bridge is None:
        return {"channels": [], "message": "Channel bridge not configured"}
    channels = bridge.list_channels()
    return {"channels": channels, "status": bridge.status().value}

channel_send

channel_send async

channel_send(request: Request)

Send a message to a channel.

Source code in src/openjarvis/server/routes.py

@router.post("/v1/channels/send")
async def channel_send(request: Request):
    """Send a message to a channel."""
    bridge = getattr(request.app.state, "channel_bridge", None)
    if bridge is None:
        raise HTTPException(status_code=503, detail="Channel bridge not configured")

    body = await request.json()
    channel_name = body.get("channel", "")
    content = body.get("content", "")
    conversation_id = body.get("conversation_id", "")

    if not channel_name or not content:
        raise HTTPException(
            status_code=400, detail="'channel' and 'content' are required",
        )

    ok = bridge.send(channel_name, content, conversation_id=conversation_id)
    if not ok:
        raise HTTPException(status_code=502, detail="Failed to send message")
    return {"status": "sent", "channel": channel_name}

channel_status

channel_status async

channel_status(request: Request)

Return channel bridge connection status.

Source code in src/openjarvis/server/routes.py

@router.get("/v1/channels/status")
async def channel_status(request: Request):
    """Return channel bridge connection status."""
    bridge = getattr(request.app.state, "channel_bridge", None)
    if bridge is None:
        return {"status": "not_configured"}
    return {"status": bridge.status().value}