Function get_app

sdk/python/feast/feature_server.py:303–799

Creates a FastAPI app that can be used to start a feature server. Args: store: The FeatureStore to use for serving features registry_ttl_sec: The TTL in seconds for the registry cache Returns: A FastAPI app Example: ```python from feast imp

(
    store: "feast.FeatureStore",
    registry_ttl_sec: int = DEFAULT_FEATURE_SERVER_REGISTRY_TTL,
)

def get_app(
    store: "feast.FeatureStore",
    registry_ttl_sec: int = DEFAULT_FEATURE_SERVER_REGISTRY_TTL,
):
    """
    Creates a FastAPI app that can be used to start a feature server.

    Args:
        store: The FeatureStore to use for serving features
        registry_ttl_sec: The TTL in seconds for the registry cache

    Returns:
        A FastAPI app

    Example:
        ```python
        from feast import FeatureStore

        store = FeatureStore(repo_path="feature_repo")
        app = get_app(store)
        ```

    The app provides the following endpoints:
    - `/get-online-features`: Get online features
    - `/retrieve-online-documents`: Retrieve online documents
    - `/push`: Push features to the feature store
    - `/write-to-online-store`: Write to the online store
    - `/health`: Health check
    - `/materialize`: Materialize features
    - `/materialize-incremental`: Materialize features incrementally
    - `/chat`: Chat UI
    - `/ws/chat`: WebSocket endpoint for chat
    MCP Support:
    - If MCP is enabled in feature server configuration, MCP endpoints will be added automatically
    """
    proto_json.patch()
    # Asynchronously refresh registry, notifying shutdown and canceling the active timer if the app is shutting down
    shutting_down = False
    active_timer: Optional[threading.Timer] = None
    # --- Offline write batching config and batcher ---
    fs_cfg = getattr(store.config, "feature_server", None)
    batching_cfg = None
    if fs_cfg is not None:
        enabled = getattr(fs_cfg, "offline_push_batching_enabled", False)
        batch_size = getattr(fs_cfg, "offline_push_batching_batch_size", None)
        batch_interval_seconds = getattr(
            fs_cfg, "offline_push_batching_batch_interval_seconds", None
        )

        if enabled is True:
            size_ok = isinstance(batch_size, int) and not isinstance(batch_size, bool)
            interval_ok = isinstance(batch_interval_seconds, int) and not isinstance(
                batch_interval_seconds, bool
            )
            if size_ok and interval_ok:
                batching_cfg = SimpleNamespace(
                    enabled=True,
                    batch_size=batch_size,