feat: inline Google Sheets data source integration

Author: cmd-err

app/ai/voice/agents/breeze_buddy/dispatch/worker.py

@@ -52,6 +52,9 @@
     _release_number,
     _run_pre_checks_for_lead,
 )
+from app.ai.voice.agents.breeze_buddy.managers.data_source_prefetch import (
+    prefetch_data_sources,
+)
 from app.ai.voice.agents.breeze_buddy.managers.utils import (
     prepare_and_store_initial_greeting,
 )
@@ -339,6 +342,8 @@ async def _dispatch(
 
             if template:
                 template = apply_playground_overrides(locked, template)
+                if template.data_sources:
+                    asyncio.create_task(prefetch_data_sources(template=template))
                 await prepare_and_store_initial_greeting(
                     lead_id=locked.id,
                     payload=locked.payload or {},

app/ai/voice/agents/breeze_buddy/managers/data_source_prefetch.py

@@ -0,0 +1,93 @@
+"""
+Data Source Prefetch Manager
+
+Pre-warms Redis with Google Sheets content for all DataSourceRefs attached to a
+template at dispatch time. Runs concurrently with greeting TTS synthesis.
+
+Cache key is scoped to a content hash so concurrent calls sharing the same sheet
+use a single cached copy.
+
+Cache key : ``datasource:{sheet_id}:{sheet_name}:{cols_hash}``
+TTL        : 60 s
+"""
+
+import asyncio
+import hashlib
+import json
+from typing import Optional
+
+from app.ai.voice.agents.breeze_buddy.template.types import DataSourceRef, TemplateModel
+from app.core.logger import logger
+from app.services.data_sources import DATA_SOURCE_UNAVAILABLE
+from app.services.google.sheets import extract_spreadsheet_id, fetch_formatted
+from app.services.redis import get_redis_service
+
+_CACHE_TTL = 60  # seconds — shared across leads; short to keep data fresh
+_FETCH_TIMEOUT = 5.0
+
+
+def _cache_key(ref: DataSourceRef) -> str:
+    """Derive a stable cache key from the sheet config."""
+    spreadsheet_id = extract_spreadsheet_id(ref.spreadsheet_url)
+    if not spreadsheet_id:
+        spreadsheet_id = "invalid_url"
+    sheet = ref.sheet_name or "_first_"
+    cols = json.dumps(sorted(ref.columns or []))
+    cols_digest = hashlib.md5(cols.encode()).hexdigest()[:8]
+    return f"datasource:{spreadsheet_id}:{sheet}:{cols_digest}:{ref.format.value}"
+
+
+async def _prefetch_one(ref: DataSourceRef) -> None:
+    if not ref.is_active:
+        return
+
+    spreadsheet_id = extract_spreadsheet_id(ref.spreadsheet_url)
+    if not spreadsheet_id:
+        logger.warning("Prefetch: invalid spreadsheet_url for '%s'", ref.name)
+        return
+
+    try:
+        content = await asyncio.wait_for(
+            fetch_formatted(
+                spreadsheet_id=spreadsheet_id,
+                sheet_name=ref.sheet_name,
+                columns=ref.columns,
+                format=ref.format.value,
+            ),
+            timeout=_FETCH_TIMEOUT,
+        )
+    except asyncio.TimeoutError:
+        logger.warning("Prefetch timeout for data source '%s'", ref.name)
+        return
+    except Exception as exc:
+        logger.error(
+            "Prefetch error for data source '%s': %s", ref.name, exc, exc_info=True
+        )
+        return
+
+    if not content or content == DATA_SOURCE_UNAVAILABLE:
+        return
+
+    cache_key = _cache_key(ref)
+    try:
+        redis = await get_redis_service()
+        await redis.setex(cache_key, content, ttl_seconds=_CACHE_TTL)
+        logger.info(
+            "Prefetched data source '%s' (%d chars, TTL=%ds)",
+            ref.name,
+            len(content),
+            _CACHE_TTL,
+        )
+    except Exception as exc:
+        logger.error("Failed to cache data source '%s': %s", ref.name, exc)
+
+
+async def prefetch_data_sources(
+    template: Optional[TemplateModel],
+) -> None:
+    if not template or not template.data_sources:
+        return
+    await asyncio.gather(
+        *[_prefetch_one(ref) for ref in template.data_sources],
+        return_exceptions=True,
+    )

app/ai/voice/agents/breeze_buddy/template/field_reference.json

@@ -17,7 +17,8 @@
     "secrets":                           { "description": "Flat dict of credential key-value pairs injected as template vars, overriding reseller-level credentials for the same keys. Values substitute into {secret_key} placeholders in HTTP URLs, headers, and bodies.", "example": { "api_token": "Bearer sk-prod-abc123", "webhook_secret": "wh_live_xyz789" } },
     "outbound_number_id":                { "description": "UUID of the outbound phone number record to use for calls from this template. Must belong to the reseller. Null to use the reseller-default number.", "example": "eb0df4e9-a438-4ce4-a828-addf9db5ce3a" },
     "is_active":                         { "description": "When false, the template is soft-disabled: dispatch will not enqueue new leads for it. Set to false (not delete) when retiring a template.", "example": true },
-    "supported_channels":                { "description": "Channels this template may be served on. Default ['voice']. Add 'chat' to enable the Buddy Assist web-chat mode. Must have at least one entry.", "example": ["voice"] }
+    "supported_channels":                { "description": "Channels this template may be served on. Default ['voice']. Add 'chat' to enable the Buddy Assist web-chat mode. Must have at least one entry.", "example": ["voice"] },
+    "data_sources":                      { "description": "Google Sheets data sources attached to this template. Each ref has name, spreadsheet_url, sheet_name, columns filter, output format (markdown_table/csv/json), and is_active flag. Fetched at call dispatch and injected as {datasource_<name>} template vars into LLM context. For example, a ref named 'customer_list' is referenced as {datasource_customer_list} in prompts.", "example": [{ "name": "customer_list", "spreadsheet_url": "https://docs.google.com/spreadsheets/d/abc123", "sheet_name": "Sheet1", "columns": ["Name", "Status"], "format": "markdown_table", "is_active": true }] }
   },
 
   "ConfigurationModel": {

app/ai/voice/agents/breeze_buddy/template/loader.py

@@ -4,12 +4,16 @@
 This module provides functionality to load templates from the database.
 """
 
+import asyncio
+import hashlib
+import json
 from typing import Dict, Optional, Tuple
 
 from app.ai.voice.agents.breeze_buddy.template.transformation_function import (
     TEMPLATE_FUNCTION_REGISTRY,
 )
 from app.ai.voice.agents.breeze_buddy.template.types import (
+    DataSourceRef,
     FlowMode,
     TemplateModel,
 )
@@ -21,6 +25,9 @@
 from app.database.accessor.breeze_buddy.template import (
     get_template_by_id_with_fallback,
 )
+from app.services.data_sources import DATA_SOURCE_UNAVAILABLE
+from app.services.google.sheets import extract_spreadsheet_id, fetch_formatted
+from app.services.redis import get_redis_service
 
 
 class FlowConfigLoader:
@@ -205,6 +212,23 @@ async def load_template(
                         f"Injected extra payload field '{field_name}' into template_vars"
                     )
 
+        # 5. data sources — inject as {datasource_<name>} variables
+        if template_obj.data_sources:
+            contents = await asyncio.gather(
+                *[
+                    self._fetch_one_data_source(ref)
+                    for ref in template_obj.data_sources
+                ],
+                return_exceptions=True,
+            )
+            for ref, result in zip(template_obj.data_sources, contents):
+                if not isinstance(result, Exception) and result is not None:
+                    var_key = f"datasource_{ref.name}"
+                    if var_key not in template_vars:
+                        template_vars[var_key] = result
+                        logger.info(
+                            f"Injected data source '{ref.name}' as variable '{var_key}'"
+                        )
         # Direct mode has a flat structure (system_prompt + flat function list)
         # rather than a nodes array, so render the top-level message fields and
         # skip the per-node loop entirely.
@@ -275,3 +299,59 @@ async def load_template(
 
         logger.info(f"Rendered task messages for template {template_obj.name}")
         return template_obj, template_vars
+
+    async def _fetch_one_data_source(self, ref: "DataSourceRef") -> str:
+        if not ref.is_active:
+            return DATA_SOURCE_UNAVAILABLE
+
+        spreadsheet_id = extract_spreadsheet_id(ref.spreadsheet_url)
+        if not spreadsheet_id:
+            logger.warning(
+                "Data source '%s' has malformed spreadsheet_url: %s",
+                ref.name,
+                ref.spreadsheet_url,
+            )
+            return DATA_SOURCE_UNAVAILABLE
+
+        sheet = ref.sheet_name or "_first_"
+        cols = json.dumps(sorted(ref.columns or []))
+        cols_digest = hashlib.md5(cols.encode()).hexdigest()[:8]
+        cache_key = (
+            f"datasource:{spreadsheet_id}:{sheet}:{cols_digest}:{ref.format.value}"
+        )
+
+        try:
+            redis = await get_redis_service()
+            cached = await redis.get(cache_key)
+            if cached:
+                return cached
+        except Exception:
+            pass
+
+        try:
+            content = await asyncio.wait_for(
+                fetch_formatted(
+                    spreadsheet_id=spreadsheet_id,
+                    sheet_name=ref.sheet_name,
+                    columns=ref.columns,
+                    format=ref.format.value,
+                ),
+                timeout=5.0,
+            )
+        except asyncio.TimeoutError:
+            logger.warning("Data source '%s' fetch timed out", ref.name)
+            return DATA_SOURCE_UNAVAILABLE
+        except Exception as exc:
+            logger.error(
+                "Data source '%s' fetch failed: %s", ref.name, exc, exc_info=True
+            )
+            return DATA_SOURCE_UNAVAILABLE
+
+        if content != DATA_SOURCE_UNAVAILABLE:
+            try:
+                redis = await get_redis_service()
+                await redis.setex(cache_key, content, ttl_seconds=60)
+            except Exception:
+                pass
+
+        return content

app/ai/voice/agents/breeze_buddy/template/types.py

@@ -2062,6 +2062,40 @@ def _default_supported_channels() -> List[Literal["voice", "chat"]]:
     return ["voice"]
 
 
+class DataSourceFormat(str, Enum):
+    MARKDOWN_TABLE = "markdown_table"
+    CSV = "csv"
+    JSON = "json"
+
+
+class DataSourceRef(BaseModel):
+    """Inline config for a data source attached to a template.
+
+    Content is fetched at call time and injected as ``template_vars[datasource_<name>]``.
+    The merchant references it in prompts with ``{datasource_<name>}`` placeholder syntax
+    (e.g. name='products' → use ``{datasource_products}`` in prompts).
+    The ``datasource_`` prefix prevents collision with payload credentials.
+    """
+
+    name: str = Field(
+        description="Variable name. Referenced in prompts as {datasource_<name>} (e.g. name='products' → {datasource_products})"
+    )
+    spreadsheet_url: str = Field(
+        description="Full Google Sheets URL (shared with platform SA)"
+    )
+    sheet_name: Optional[str] = Field(
+        default=None, description="Tab name. NULL = first tab"
+    )
+    columns: Optional[List[str]] = Field(
+        default=None, description="Columns to include. NULL = all columns"
+    )
+    format: DataSourceFormat = Field(
+        default=DataSourceFormat.MARKDOWN_TABLE,
+        description="Output format: 'markdown_table' | 'csv' | 'json'",
+    )
+    is_active: bool = Field(default=True, description="Skip if false at call time")
+
+
 class TemplateModel(BaseModel):
     # Read-only fields (set by server, not editable via API).
     # These are intentionally excluded from ReplaceTemplateRequest so that
@@ -2081,6 +2115,10 @@ class TemplateModel(BaseModel):
     secrets: Optional[Dict[str, Any]] = None
     outbound_number_id: Optional[str] = None
     is_active: bool = True
+    data_sources: Optional[List[DataSourceRef]] = Field(
+        default=None,
+        description="Inline data source configs attached to this template",
+    )
     # Channels this template is allowed to be served on. Defaults to
     # voice-only so existing templates are unaffected. Add "chat" to
     # opt the template into the chat (text) mode flow build path
@@ -2129,6 +2167,7 @@ class CreateTemplateRequest(BaseModel):
     expected_callback_response_schema: Optional[Dict[str, Any]] = None
     configurations: Optional[ConfigurationModel] = None
     secrets: Optional[Dict[str, Any]] = None
+    data_sources: Optional[List[DataSourceRef]] = None
     supported_channels: List[Literal["voice", "chat"]] = Field(
         default_factory=_default_supported_channels,
         min_length=1,
@@ -2177,6 +2216,7 @@ class ReplaceTemplateRequest(BaseModel):
     expected_callback_response_schema: Optional[Dict[str, Any]] = None
     configurations: Optional[ConfigurationModel] = None
     secrets: Optional[Dict[str, Any]] = None
+    data_sources: Optional[List[DataSourceRef]] = None
     supported_channels: Optional[List[Literal["voice", "chat"]]] = Field(
         default=None,
         min_length=1,

app/api/routers/breeze_buddy/__init__.py

@@ -15,6 +15,7 @@
 
 # Daily transport (web/mobile clients via Daily.co)
 from app.api.routers.breeze_buddy.daily import router as daily_router
+from app.api.routers.breeze_buddy.data_sources import router as data_sources_router
 from app.api.routers.breeze_buddy.demo import router as demo_router
 from app.api.routers.breeze_buddy.leads import router as leads_router
 from app.api.routers.breeze_buddy.merchants import router as merchants_router
@@ -44,6 +45,7 @@
 # ============================================================================
 
 # Public demo (unauthenticated)
+router.include_router(data_sources_router, prefix="", tags=["data-sources"])
 router.include_router(demo_router, prefix="", tags=["demo"])
 
 # Authentication (JWT & S2S tokens)

app/api/routers/breeze_buddy/data_sources/__init__.py

@@ -0,0 +1,65 @@
+"""
+Sheet discovery endpoints (admin-only utility).
+
+Discovery routes declared BEFORE /{id} to avoid path conflict if CRUD endpoints
+are added later.
+"""
+
+from typing import List, Optional
+
+from fastapi import APIRouter, Depends, Query
+
+from app.api.security.breeze_buddy.rbac_token import get_current_user_with_rbac
+from app.core.security.authorization import require_admin
+from app.schemas import UserInfo
+from app.schemas.breeze_buddy.data_source import (
+    ColumnsResponse,
+    PreviewResponse,
+    TabsResponse,
+)
+
+from .handlers import (
+    list_columns_handler,
+    list_tabs_handler,
+    preview_handler,
+)
+
+router = APIRouter()
+
+
+@router.get("/data-sources/sheets/tabs", response_model=TabsResponse)
+async def get_sheet_tabs(
+    spreadsheet_url: str = Query(..., description="Full Google Sheets URL"),
+    current_user: UserInfo = Depends(get_current_user_with_rbac),
+):
+    """List all tab names in a Google Spreadsheet (admin-only)."""
+    require_admin(current_user)
+    return await list_tabs_handler(spreadsheet_url, current_user)
+
+
+@router.get("/data-sources/sheets/columns", response_model=ColumnsResponse)
+async def get_sheet_columns(
+    spreadsheet_url: str = Query(..., description="Full Google Sheets URL"),
+    sheet_name: Optional[str] = Query(
+        None, description="Tab name (default: first tab)"
+    ),
+    current_user: UserInfo = Depends(get_current_user_with_rbac),
+):
+    """List column headers for a sheet tab (admin-only)."""
+    require_admin(current_user)
+    return await list_columns_handler(spreadsheet_url, sheet_name)
+
+
+@router.get("/data-sources/sheets/preview", response_model=PreviewResponse)
+async def preview_sheet(
+    spreadsheet_url: str = Query(..., description="Full Google Sheets URL"),
+    sheet_name: Optional[str] = Query(
+        None, description="Tab name (default: first tab)"
+    ),
+    columns: Optional[List[str]] = Query(None, description="Columns to include"),
+    max_rows: int = Query(10, ge=1, le=100, description="Max rows to return"),
+    current_user: UserInfo = Depends(get_current_user_with_rbac),
+):
+    """Preview up to N rows from a sheet (admin-only)."""
+    require_admin(current_user)
+    return await preview_handler(spreadsheet_url, sheet_name, columns, max_rows)