Method get_fast_api_app

src/google/adk/cli/api_server.py:872–1030

Creates a FastAPI app for the ADK web server. By default it'll just return a FastAPI instance with the API server endpoints, but if you specify a web_assets_dir, it'll also serve the static web assets from that directory. Args: lifespan: The lifespan of the FastAPI app.

(
      self,
      lifespan: Optional[Lifespan[FastAPI]] = None,
      allow_origins: Optional[list[str]] = None,
      web_assets_dir: Optional[str] = None,
      setup_observer: Callable[
          [Observer, "ApiServer"], None
      ] = lambda o, s: None,
      tear_down_observer: Callable[
          [Observer, "ApiServer"], None
      ] = lambda o, s: None,
      register_processors: Callable[[TracerProvider], None] = lambda o: None,
      otel_to_cloud: bool = False,
      with_ui: bool = False,
  )

      raise HTTPException(status_code=500, detail=str(e)) from e

  def get_fast_api_app(
      self,
      lifespan: Optional[Lifespan[FastAPI]] = None,
      allow_origins: Optional[list[str]] = None,
      web_assets_dir: Optional[str] = None,
      setup_observer: Callable[
          [Observer, "ApiServer"], None
      ] = lambda o, s: None,
      tear_down_observer: Callable[
          [Observer, "ApiServer"], None
      ] = lambda o, s: None,
      register_processors: Callable[[TracerProvider], None] = lambda o: None,
      otel_to_cloud: bool = False,
      with_ui: bool = False,
  ):
    """Creates a FastAPI app for the ADK web server.

    By default it'll just return a FastAPI instance with the API server
    endpoints,
    but if you specify a web_assets_dir, it'll also serve the static web assets
    from that directory.

    Args:
      lifespan: The lifespan of the FastAPI app.
      allow_origins: The origins that are allowed to make cross-origin requests.
        Entries can be literal origins (e.g., 'https://example.com') or regex
        patterns prefixed with 'regex:' (e.g.,
        'regex:https://.*\\.example\\.com').
      web_assets_dir: The directory containing the web assets to serve.
      setup_observer: Callback for setting up the file system observer.
      tear_down_observer: Callback for cleaning up the file system observer.
      register_processors: Callback for additional Span processors to be added
        to the TracerProvider.
      otel_to_cloud: Whether to enable Cloud Trace and Cloud Logging
        integrations.

    Returns:
      A FastAPI app instance.
    """
    trace_dict = {}
    session_trace_dict = {}
    self._trace_dict = trace_dict
    self._session_trace_dict = session_trace_dict

    # Set up a file system watcher to detect changes in the agents directory.
    observer = Observer()
    setup_observer(observer, self)

    @asynccontextmanager
    async def internal_lifespan(app: FastAPI):
      try:
        if lifespan:
          async with lifespan(app) as lifespan_context:
            yield lifespan_context
        else:
          yield
      finally:
        tear_down_observer(observer, self)