The JARVIS backend is built with FastAPI and follows a modular pipeline architecture for processing captured media, identifying faces, enriching person data, and synthesizing intelligence reports.
The application uses FastAPI’s lifespan context manager for startup/shutdown:
backend/main.py
@asynccontextmanagerasync def lifespan(app: FastAPI) -> AsyncIterator[None]: logger.info( "JARVIS started — det={} emb={} db={} face_search={} exa={} deep_researcher={} synth={} (primary) synth_fallback={} supermemory={}", detector.__class__.__name__, embedder.__class__.__name__, db_gateway.__class__.__name__, face_searcher is not None, exa_client is not None, deep_researcher is not None, synthesis_engine.__class__.__name__ if synthesis_engine else None, synthesis_fallback.__class__.__name__ if synthesis_fallback else None, supermemory_client is not None, ) if telegram_bot: await telegram_bot.start() yield if telegram_bot: await telegram_bot.stop() if supermemory_client: await supermemory_client.close() logger.info("JARVIS shutting down")
Returns configuration status for all integrated services:
backend/main.py
@app.get("/api/services", response_model=list[ServiceStatus])async def services() -> list[ServiceStatus]: descriptions = { "convex": "Real-time board subscriptions and mutations", "mongodb": "Persistent raw captures and dossiers", "exa": "Fast pass research and person lookup", "browser_use": "Deep research browser agents", "openai": "Transcription and fallback LLM integrations", "anthropic": "Primary synthesis model (Claude)", "gemini": "Fallback vision and synthesis model when Anthropic unavailable", "laminar": "Tracing and evaluation telemetry", "telegram": "Glasses-side media intake", "pimeyes_pool": "Rotating account pool for identification", } flags = settings.service_flags() return [ ServiceStatus(name=name, configured=configured, notes=descriptions.get(name)) for name, configured in flags.items() ]
The backend exposes which services are configured via environment variables:
# Check if services are availableflags = settings.service_flags()if flags["browser_use"]: # Deep research available orchestrator = ResearchOrchestrator(settings)