Function query

src/google/adk/cli/fast_api.py:847–883

(request: Request)

        response_class=JSONResponse,
    )
    async def 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 {})

      try:
        json_serialized_content = jsonable_encoder({"output": output})
      except ValueError as encoding_error:
        logging.exception(
            "FastAPI could not JSON-encode the response from invocation method"
            " %s. Error: %s. Invocation method's original response: %r",
            parsed.class_method,
            encoding_error,
            output,
        )
        raise
      return JSONResponse(content=json_serialized_content)

    @app.post(
        "/api/stream_reasoning_engine",