Relay

RELAY is an autonomous multi-agent system that monitors FDA drug shortages to find safe substitutes, create pharmacist-ready protocols, and continuously learn from past responses.

Comment

RELAY — Every Drug Shortage. Solved in 60 Seconds. Powered by Elasticsearch.

From Idea to Impact

The Problem — A hospital pharmacy has 1,600+ active drug shortages and no intelligence layer to handle them

Every hospital pharmacy in the United States faces the same weekly crisis: FDA drug shortages. Not occasional, not rare — permanent. The FDA currently tracks over 1,600 active shortages, and when a drug goes short, a pharmacist is expected to:

  1. Verify the shortage is real and still active (checking FDA, ASHP, and manufacturer sites manually)
  2. Find a clinically appropriate substitute — one that matches indication, route, mechanism, and contraindications
  3. Screen every candidate for safety flags — boxed warnings, allergen conflicts, route mismatches
  4. Cross-reference what their own institution chose last time this drug was short
  5. Ground the recommendation in current clinical guidelines and web evidence
  6. Write a formal protocol for pharmacy and medical staff

This takes 30–90 minutes per drug. Multiplied across 1,600+ active shortages. And the institutional knowledge generated — which drugs worked, which were excluded and why — is almost never systematically retained. The next time the same drug goes on shortage (and it will), the analysis starts from scratch.

The gap The cost
Manual substitute search 30–90 min per shortage, duplicated across every hospital in the country
Keyword search misses clinical equivalents "Methotrexate" doesn't find "DMARD" or "folate antagonist" drugs
Institutional memory evaporates Same analysis repeated every shortage cycle
No real-time grounding Protocols written without verifying current FDA/ASHP status
No severity triage or alerting All 1,600 shortages look the same — no signal on what's most critical

The Solution — RELAY: an autonomous pipeline that handles every shortage in under 60 seconds

RELAY is a real-time drug shortage intelligence system built on Elasticsearch ELSER semantic search, Hybrid RRF, ES|QL, Google ADK 2.0, the Elastic MCP server, and Gemini grounding. It runs a full seven-stage pipeline on every shortage and streams the results live to a pharmacist dashboard.

The pharmacist selects a shortage from the live feed, clicks Run Pipeline, and watches:

  1. WATCHER confirms the shortage is active in the FDA database
  2. IMPACT scores its clinical severity (0–1 scale based on breadth, therapeutic class, and recency)
  3. SUBSTITUTE — an ADK 2.0 LlmAgent calls the Elastic MCP server to run a Hybrid RRF query combining ELSER semantic search and BM25 over FDA drug labels, finding candidates that match on clinical meaning — visible on every candidate card as a "Hybrid RRF" badge
  4. SAFETY screens every candidate for boxed warnings, allergen conflicts, and route mismatches
  5. MEMORY — ADK 2.0 + Elastic MCP recalls the top 5 most relevant past decisions, then saves the new decision when done
  6. PROTOCOL compiles everything into a structured pharmacist-ready document
  7. GEMINI GROUNDING hits live web sources to verify shortage status and cite clinical guidelines, returning real citations from FDA, ASHP, and NIH

The output is a downloadable PDF protocol. Every stage streams to the browser in real time via Server-Sent Events. The whole pipeline takes under 60 seconds.

What It Does

Stage / Feature What happens Technology
WATCHER Polls openFDA every 15 min, deduplicates by field collapse, bulk-indexes new shortages APScheduler + Elasticsearch field collapse
IMPACT Severity score: breadth × therapeutic class × recency Python scoring logic
SUBSTITUTE ADK 2.0 LlmAgent calls Elastic MCP with Hybrid RRF: ELSER + BM25 Google ADK 2.0 + Elastic MCP + ELSER + Hybrid RRF
SAFETY Boxed warning screen, allergen conflict check, route mismatch flag openFDA label data
MEMORY ADK 2.0 LlmAgent calls Elastic MCP search (recall) + index_document (save) Google ADK 2.0 + Elastic MCP + ELSER
PROTOCOL Structured pharmacist document with ranked candidates and safety flags FastAPI SSE
GROUNDING Live FDA/ASHP/NIH verification with real citations Gemini 3 Flash + Vertex AI GoogleSearch
Analytics Dashboard Bar charts, severity histogram, stat cards from ES aggregations Elasticsearch Aggregations
**ES\ QL Insights** 4 live insight cards powered by ES\
Real-time Alerting + Slack ES severity monitor fires on score ≥ 0.8 — dismissible red banner + Slack Block Kit notification; ES-backed dedup prevents duplicates Elasticsearch + APScheduler + Slack Webhooks
Intelligence Brief PDF One-click PDF export: exec summary, ES\ QL insights, drug severity table, class bars

How Elasticsearch Is Central — Not Peripheral

RELAY is not a system that uses Elasticsearch for storage and does intelligence elsewhere. Every intelligent step runs through Elasticsearch. Five distinct Elasticsearch capabilities are used in production:

1. Hybrid BM25 + ELSER Reciprocal Rank Fusion (RRF)

The SUBSTITUTE stage uses a dual sub-search with RRF to get the best of both retrieval approaches:

{
  "sub_searches": [
    { "query": { "semantic": { "field": "indications", "query": "<full indication context>" } } },
    { "query": { "multi_match": { "query": "<drug name + key terms>", "fields": ["indications", "drug_name^2", "generic_name^2"], "type": "best_fields" } } }
  ],
  "rank": { "rrf": { "window_size": 50, "rank_constant": 20 } }
}

ELSER catches semantically equivalent drugs that don't share keywords. BM25 catches exact drug name and indication term matches. RRF merges both ranked lists without requiring score normalization. The search method is visible to users as the "Hybrid RRF" badge on every substitute candidate card.

2. ES|QL Live Insights

The Analytics page runs four ES|QL queries on every load, powering four real-time insight cards. Each card shows the actual ES|QL query text so the query-to-insight path is fully transparent:

FROM relay-shortages
| WHERE status == "current" AND severity_score >= 0.8
| STATS critical_drugs = COUNT_DISTINCT(drug_name)

FROM relay-shortages
| WHERE status == "current" AND therapeutic_class != ""
| STATS avg_sev = AVG(severity_score) BY therapeutic_class
| SORT avg_sev DESC | LIMIT 1

FROM relay-shortages
| WHERE status == "current"
| STATS max_sev = MAX(severity_score), avg_sev = AVG(severity_score)

FROM relay-shortages
| WHERE status == "resolved"
| STATS resolved_drugs = COUNT_DISTINCT(drug_name)

Every card carries an "⚡ ES|QL" badge and the full query text — making the analytics provenance completely visible.

3. Elasticsearch Aggregations — Analytics Dashboard

The /api/analytics endpoint runs five aggregations in a single Elasticsearch query, with size=0 so Elasticsearch does all the work server-side:

aggs={
    "by_class":           {"terms": {"field": "therapeutic_class", "size": 10, "order": {"_count": "desc"}}},
    "top_by_severity":    {"terms": {"field": "drug_name", "size": 10, "order": {"max_sev": "desc"}}, "aggs": {"max_sev": {"max": {"field": "severity_score"}}}},
    "severity_histogram": {"histogram": {"field": "severity_score", "interval": 0.2}},
    "avg_severity":       {"avg": {"field": "severity_score"}},
    "unique_drugs":       {"cardinality": {"field": "drug_name"}},
}

These power the bar charts, severity distribution histogram, and stat cards on the Analytics page — a fully Elasticsearch-driven analytics view.

4. Real-time Severity Alerting + Slack Notifications via Elasticsearch Monitor

An APScheduler job runs every 60 seconds — querying Elasticsearch for drugs at or above severity_score >= 0.8 using field_collapse for deduplication, then firing alerts for any drug not previously seen:

result = await es.search(
    index=RELAY_SHORTAGES,
    query={"bool": {"must": [
        {"term": {"status": "current"}},
        {"range": {"severity_score": {"gte": 0.8}}},
    ]}},
    collapse={"field": "drug_name"},
    sort=[{"severity_score": "desc"}],
    size=10,
)

New critical drugs appear immediately as a live dismissible red banner at the top of the dashboard with CRITICAL / HIGH / ELEVATED severity badges. Each alert shows the drug name, severity, time elapsed, and a one-click "Run Pipeline" shortcut.

For each new critical drug, a formatted Slack Block Kit message is posted to a configured incoming webhook — drug name, severity score, alert tier (CRITICAL / HIGH / ELEVATED), and a direct link to the RELAY dashboard.

Deduplication across Cloud Run instances is backed by Elasticsearch: before sending, the system atomically calls es.create() with a deterministic document ID (slack-{date}-{drug}). If the document already exists, Elasticsearch returns a 409 Conflict and the notification is skipped. This means only one instance can ever "win" the send for a given drug per day — no duplicate Slack messages regardless of how many Cloud Run instances are running or how often the service restarts.

# Atomic send-slot claim — cross-instance dedup via ES create
await es.create(
    index="relay-alert-log",
    id=f"slack-{today}-{safe_drug_name}",
    document={"drug_name": drug_name, "sent_at": datetime.now(timezone.utc).isoformat()},
)

5. Field Collapse for Intelligent Feed Deduplication

The live shortage feed uses Elasticsearch field_collapse on drug_name to return one entry per unique drug, regardless of how many FDA source records exist for it:

result = await es.search(
    index=RELAY_SHORTAGES,
    query={"term": {"status": "current"}},
    collapse={"field": "drug_name"},
    size=500,
)

The feed header shows the deduplicated count pulled directly from this query. This gives pharmacists a clean, actionable view of 500 unique drugs in shortage rather than thousands of duplicated records.

The Elastic MCP Server as the Intelligence Boundary

All SUBSTITUTE searches and MEMORY operations go through the official Elastic MCP server (@elastic/mcp-server-elasticsearch) via ADK 2.0 MCPToolset.from_server() (stdio subprocess). The ADK agent calls:

  • search with a Hybrid RRF query body → substitute candidates / memory recall
  • index_document → memory save after each protocol

The intelligence layer (ADK + Gemini) and the data layer (Elasticsearch) are joined at the MCP protocol boundary.

How It Was Built

Elasticsearch (Elastic Cloud)

Four indices power the entire system:

Index Purpose Key fields
relay-shortages 1,645+ active shortage records from openFDA drug_name, status, severity_score, therapeutic_class
relay-labels Drug label documents (ELSER indexed) indications, contraindications, warnings, dosage_text (semantic_text)
relay-substitutions Runtime substitute records candidate, semantic_match, safety_status
relay-response-memory Institutional memory pattern_summary (semantic_text), chosen_substitute, excluded, rationale

The semantic_text field type triggers ELSER inference automatically on index and on query — no separate embedding step, no external model call.

Backend — FastAPI + Google ADK 2.0 + Elastic MCP

The pipeline runs as an async generator. Each stage yields SSE events to the browser. The SUBSTITUTE and MEMORY stages use ADK 2.0 LlmAgent connected to the Elastic MCP server:

# ADK 2.0 + Elastic MCP — SUBSTITUTE and MEMORY agents
tools, exit_stack = await MCPToolset.from_server(
    connection_params=StdioServerParameters(
        command="npx",
        args=["@elastic/mcp-server-elasticsearch"],
        env={"ES_URL": ..., "ES_API_KEY": ...},
    )
)

agent = LlmAgent(
    model="gemini-2.0-flash",
    name="relay_substitute_agent",
    instruction=HYBRID_RRF_PROMPT,  # instructs agent to use sub_searches + rank.rrf
    tools=tools,
)
runner = Runner(agent=agent, app_name="relay-substitute", session_service=InMemorySessionService())

The GROUNDING stage uses the google-genai SDK directly with Vertex AI's native GoogleSearch grounding — because gemini-3-flash-preview does not support function-calling style tools. Vertex AI handles the web retrieval server-side, so no function calling is needed on the model.

# google-genai SDK — GROUNDING agent only
response = await client.aio.models.generate_content(
    model="gemini-3-flash-preview",
    contents=prompt,
    config=types.GenerateContentConfig(
        tools=[types.Tool(google_search=types.GoogleSearch())],
    ),
)

This gives RELAY a clean split: ADK 2.0 for the clinical intelligence pipeline (drug-to-drug decisions via Elastic MCP + ELSER), google-genai SDK for live web grounding.

Frontend — React + TypeScript + SSE

The dashboard is a three-panel layout:

  • Left: Live shortage feed (500+ unique drugs, severity chips, drug name search)
  • Center: Pipeline hero — five stage nodes that pulse, turn green, and animate as each stage completes
  • Right: Protocol panel — Memory Recall cards showing prior decisions, Gemini Grounding panel with search queries and web citations, Hybrid RRF badge on substitute candidates, and a PDF download button
  • Top: Live alert banner — red dismissible banner with CRITICAL/HIGH/ELEVATED severity badges, fired by Elasticsearch severity monitor
  • Analytics page: ES aggregation charts, ES|QL insight cards, and a one-click "Download Brief" button that exports a complete Shortage Intelligence Brief PDF

Features

Feature Details
Live shortage feed 1,645+ active FDA shortages, polled every 15 min, ES field collapse for deduplication
Hybrid RRF substitution ELSER semantic + BM25 combined via Reciprocal Rank Fusion — "Hybrid RRF" badge visible
Elastic MCP integration ADK 2.0 agent calls search + index_document via official Elastic MCP server
ELSER semantic search semantic_text fields on drug labels — clinical meaning, not keyword matching
**ES\ QL live insights**
ES aggregations analytics Therapeutic class bars, severity histogram, drug severity table — all ES aggregations
Real-time ES alerting + Slack Elasticsearch severity monitor (1-min poll), dismissible red banner + Slack Block Kit notifications; ES atomic dedup across instances
Shortage Intelligence Brief One-click PDF: exec summary, ES\
Severity scoring 0–1 score: breadth × therapeutic class criticality × recency
Safety screening Boxed warnings, allergen class conflicts, route mismatches — SAFE / CAUTION / EXCLUDED
Institutional memory ELSER recall of past decisions + save new decisions — learning loop via Elastic MCP
Gemini 3 Flash grounding Live web citations from FDA, ASHP, NIH via Vertex AI native GoogleSearch
Real-time SSE streaming Every pipeline stage streams to the browser as it completes
Protocol PDF download Formatted PDF with header, ranked candidates, safety flags, citations
Cloud Run + Cloud Build Push to main → deployed in ~4 minutes

Google Cloud Services Used

Service Purpose
Vertex AI + Gemini 3 Flash Preview Native GoogleSearch grounding tool; performs live web searches at inference time for real-time FDA/ASHP/NIH citation generation
Cloud Run Serverless container hosting for FastAPI backend and React frontend
Cloud Build CI/CD pipeline — builds and deploys on every push to main
Artifact Registry Stores versioned Docker images for the API and frontend containers
Secret Manager Stores Elasticsearch URL and API key; secrets mounted at runtime, never in source control

Challenges

Integrating the Elastic MCP server into an async FastAPI pipeline

The Elastic MCP server runs as a Node.js subprocess (stdio transport). Launching it per pipeline request via MCPToolset.from_server() required Node.js 20 inside the Python Docker image (installed via NodeSource) and the MCP server pre-installed globally (npm install -g @elastic/mcp-server-elasticsearch) to avoid npx download latency at request time. The async lifecycle — starting the subprocess, running the ADK agent, closing the exit stack — had to be managed correctly inside FastAPI's async event loop without leaking processes. A direct ES client fallback was added so the pipeline stays functional if the MCP path fails.

Teaching the ADK agent to issue Hybrid RRF queries

Getting the ADK LlmAgent to consistently emit a sub_searches + rank.rrf query body (rather than a plain semantic query) required precise instruction engineering. The system prompt had to include the exact JSON structure with field names, types, and RRF parameters — because any deviation would silently fall back to a simpler query. The fallback path in the direct ES client also implements Hybrid RRF natively, so both paths produce identical search behavior.

ADK google_search tool vs. Vertex AI native grounding — and why the split exists

Google ADK's google_search built-in tool uses function calling, which gemini-3-flash-preview does not support. Vertex AI's native GoogleSearch grounding is server-side — the model doesn't call a function, Vertex AI retrieves the web context before inference. The google-genai SDK exposes this via types.Tool(google_search=types.GoogleSearch()). The result: ADK 2.0 handles the clinical pipeline (SUBSTITUTE + MEMORY via Elastic MCP with gemini-2.0-flash), and the google-genai SDK handles grounding (with gemini-3-flash-preview). Both Gemini models. Two different tool patterns. Clean split.

Gemini 3 Flash Preview — global endpoint only

The model exists in Vertex AI's Model Garden but is only available on the global endpoint, not regional ones like us-central1. The fix was setting GOOGLE_CLOUD_LOCATION=global in agent.py before each grounding call, while keeping GOOGLE_CLOUD_LOCATION=us-central1 for the ADK agents.

ELSER on a fresh trial cluster

On a new Elastic Cloud trial, ELSER inference takes time to warm up. The first indexing run for drug labels returned 0 documents visible until forced refresh. The fix was to add index.refresh_interval awareness in the ingestion script and use async_bulk with explicit acknowledgment.

pip dependency resolution in Docker — twice

First round (ADK 1.4.2): google-adk>=1.0.0 triggered pip's backtracking resolver across 50+ versions because of pydantic version conflicts. Fixed by exact-pinning all major packages. Second round (ADK 2.0 upgrade): ADK 2.0 requires google-genai>=1.72,<2 and pydantic>=2.12. Fixed by: google-adk[mcp]==2.0.0, google-genai==1.75.0, pydantic==2.12.0, fastapi==0.124.1.

nginx HTTPS proxy on Cloud Run

The frontend nginx container proxied to http://backend:8000 (Docker Compose style) which doesn't exist on Cloud Run. The fix was injecting BACKEND_URL at container startup via envsubst, and adding proxy_ssl_server_name on + proxy_set_header Host $proxy_host so nginx correctly handles the TLS handshake and Cloud Run routing for HTTPS upstream.

Accomplishments

  • Built a fully working real-time clinical intelligence pipeline that runs from raw shortage data to grounded, cited, downloadable protocol in under 60 seconds
  • Five distinct Elasticsearch capabilities in production: ELSER semantic search, Hybrid RRF, ES|QL analytics, ES aggregations dashboard, real-time threshold alerting — all visible and demonstrable
  • The Elastic MCP server is genuinely integrated — ADK 2.0 agents call search and index_document via the MCP protocol for all substitute and memory operations
  • Hybrid RRF finding clinically meaningful substitutes — ELSER catches mechanism equivalents, BM25 catches name matches, RRF combines both — and the search method is visible on every card
  • ES|QL insight cards showing real-time analytics with the actual query text on screen — the query-to-insight path is fully transparent
  • Gemini grounding surfacing real-time safety flags: during testing, the grounding layer identified a May 2026 FDA voluntary withdrawal of Tazverik (tazemetostat) for secondary malignancy risk — the kind of information that would be missed by any static database
  • Institutional memory that actually compounds: the ELSER recall correctly surfaces prior methotrexate decisions when a new methotrexate formulation shortage comes in, even with different drug name spelling
  • Slack alerting with ES-backed dedup — the severity monitor fires Slack Block Kit messages to #pharmacy-ops the moment a drug crosses the critical threshold; atomic es.create() prevents duplicate notifications across Cloud Run instances
  • Full Cloud Run deployment with automated Cloud Build CI/CD — push to main, both services live in under 4 minutes

What's Next

  • Therapeutic class filtering — Add a pre-filter on Hybrid RRF results to restrict substitution search to the same or closely related therapeutic class
  • WATCHER alert escalation — Route critical alerts to on-call paging systems (PagerDuty, OpsGenie) beyond the current Slack notification
  • Institution-level memory isolation — Namespace memory records by hospital/institution so that memory recall reflects your institution's decisions, not a shared pool
  • EHR integration — Connect to FHIR endpoints to cross-check substitute candidates against formulary and active patient allergies in real time
  • Kibana dashboards — Embed live Kibana Lens visualizations directly in the Analytics page for judges and users on the same Elastic Cloud cluster

Built With

Share this project:

Updates