app

app

FastAPI application factory for the OpenJarvis API server.

Functions

create_app

create_app(engine, model: str, *, agent=None, bus=None, engine_name: str = '', agent_name: str = '', channel_bridge=None, config=None, speech_backend=None, agent_manager=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,
    speech_backend=None,
    agent_manager=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 as exc:
            logger.debug("Security guardrails init skipped: %s", exc)

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

    from fastapi.middleware.cors import CORSMiddleware
    app.add_middleware(
        CORSMiddleware,
        allow_origins=["*"],
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )

    # 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.speech_backend = speech_backend
    app.state.agent_manager = agent_manager
    app.state.session_start = time.time()

    app.include_router(router)
    app.include_router(dashboard_router)
    app.include_router(comparison_router)
    include_all_routes(app)

    # Add security headers middleware
    try:
        from openjarvis.server.middleware import create_security_middleware

        middleware_cls = create_security_middleware()
        if middleware_cls is not None:
            app.add_middleware(middleware_cls)
    except Exception as exc:
        logger.debug("Security middleware init skipped: %s", exc)

    # 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 static files directly, fall back to index.html for SPA routes."""
            if full_path:
                candidate = (static_dir / full_path).resolve()
                # Path traversal prevention
                resolved_root = static_dir.resolve()
                if candidate.is_relative_to(resolved_root) and candidate.is_file():
                    return FileResponse(candidate, headers=_NO_CACHE_HEADERS)
            return FileResponse(
                static_dir / "index.html",
                headers=_NO_CACHE_HEADERS,
            )

    return app