research_router

research_router

HTTP route: POST /api/research — agentic research over the knowledge store.

Drives :class:openjarvis.agents.research_loop.ResearchAgent and streams a custom SSE event schema back to the client:

• search_call — about to invoke HybridSearch.search (with arguments)
• search_result — search returned (num_hits, top_titles)
• synthesis — final answer, emitted in word-window chunks for an incremental UX (the agent itself returns the full string in one shot; chunking happens in the router so we don't need to rewire the loop)
• done — sentinel marking the end of the stream

Clarify is disabled for the web session — the agent's clarify_handler is overridden to return a fixed "no clarification available" string so the loop never blocks waiting for terminal stdin. Bringing real clarify back to the browser will require a two-step session protocol; that's a future endpoint, not this one.

Classes

Functions

research async

research(req: ResearchRequest, request: Request) -> StreamingResponse

Run a research query and stream the agent's trace + synthesis via SSE.

Response is text/event-stream with one JSON event per frame. See the module docstring for the schema; a final {"type": "done"} always terminates the stream so clients can detect end-of-response without parsing the underlying [DONE] sentinel used by OpenAI-style routes.

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

@router.post("/research")
async def research(req: ResearchRequest, request: Request) -> StreamingResponse:
    """Run a research query and stream the agent's trace + synthesis via SSE.

    Response is ``text/event-stream`` with one JSON event per frame. See the
    module docstring for the schema; a final ``{"type": "done"}`` always
    terminates the stream so clients can detect end-of-response without
    parsing the underlying ``[DONE]`` sentinel used by OpenAI-style routes.
    """
    active_engine = getattr(request.app.state, "engine", None)
    active_model = str(getattr(request.app.state, "model", "") or "")
    active_engine_key = str(getattr(request.app.state, "engine_name", "") or "")
    if active_engine is not None and not active_engine_key:
        active_engine_key = str(getattr(active_engine, "engine_id", "") or "")
    return StreamingResponse(
        _stream_research(
            req.query,
            active_engine=active_engine,
            active_engine_key=active_engine_key,
            active_model=active_model,
            request_model=req.model or "",
        ),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",
        },
    )