Function stream_query

src/google/adk/cli/fast_api.py:890–932

(request: Request)

        response_class=StreamingResponse,
    )
    async def stream_query(request: Request):
      try:
        body = await request.json()
      except json.JSONDecodeError as exc:
        raise HTTPException(status_code=400, detail=f"Invalid JSON: {exc}")
      try:
        parsed = _QueryRequest.model_validate(body)
      except _ValidationError as exc:
        raise HTTPException(status_code=400, detail=exc.errors())
      if not adk_app._tmpl_attrs.get("runner"):
        adk_app._tmpl_attrs["runner"] = await adk_web_server.get_runner_async(
            app_name=gemini_enterprise_app_name
        )
      if parsed.class_method is None:
        raise HTTPException(
            status_code=400, detail="class_method cannot be None"
        )
      if parsed.class_method not in _ALLOWED_AGENT_ENGINE_CLASS_METHODS:
        raise HTTPException(
            status_code=400,
            detail=f"class_method {parsed.class_method} is not allowed",
        )
      method = getattr(adk_app, parsed.class_method)
      output = await _invoke_callable_or_raise(method, parsed.input or {})

      if inspect.isgenerator(output):

        async def _aiter_from_iter(iterator):
          while True:
            try:
              chunk = await run_in_threadpool(next, iterator)
              yield chunk
            except StopIteration:
              break

        content_iter = _aiter_from_iter(output)
      else:
        content_iter = output

      return StreamingResponse(
          content=json_generator(content_iter),
          media_type="application/json",
      )

  return app