feat: Use Vertex AI instead of Gemini

.github/workflows/deploy.yml

@@ -12,6 +12,7 @@ jobs:
     runs-on: ubuntu-latest
     permissions:
       contents: write # Required for gh-pages deployment
+      id-token: write # Required for Workload Identity Federation
 
     steps:
       - uses: actions/checkout@v5
@@ -24,9 +25,17 @@ jobs:
       - name: Install dependencies
         run: uv sync
 
+      - id: auth
+        uses: google-github-actions/auth@v2
+        with:
+          workload_identity_provider: ${{ secrets.GCP_WIF_PROVIDER }}
+          service_account: ${{ secrets.GCP_SA_EMAIL }}
+
       - name: Generate static HTML
         env:
-          GEMINI_API_KEY: ${{ secrets.GEMINI_API_KEY }}
+          GOOGLE_CLOUD_PROJECT: ${{ secrets.GCP_PROJECT_ID }}
+          GOOGLE_CLOUD_LOCATION: global
+          AI_MODEL: claude-sonnet-4-6
         run: |
           mkdir -p data
           uv run cfp-radar collect --no-ai data/index.html

README.md

@@ -18,10 +18,18 @@ curl -LsSf https://astral.sh/uv/install.sh | sh
 # Install dependencies
 uv sync
 
-# Set required environment variables
-export GEMINI_API_KEY="your-key"
+# Set required environment variables for Claude on Vertex AI
+export GOOGLE_CLOUD_PROJECT="your-project"
+# Or: export ANTHROPIC_VERTEX_PROJECT_ID="your-project"
+export GOOGLE_CLOUD_LOCATION="global"
+export AI_MODEL="claude-sonnet-4-6"
+gcloud auth application-default login
 ```
 
+You need the `roles/aiplatform.user` role on the GCP project. Claude web search also requires the `constraints/vertexai.allowedPartnerModelFeatures` org policy to allow web search. Collection works without GCP credentials when using `--no-ai`.
+
+See [Claude on Vertex AI](https://docs.anthropic.com/en/api/claude-on-vertex-ai) and [Vertex AI pricing](https://cloud.google.com/vertex-ai/generative-ai/pricing) for details.
+
 Set `export SLACK_WEBHOOK_URL="your-webhook-url"` if you plan to use Slack notifications.
 
 ## Configuration

pyproject.toml

@@ -5,7 +5,7 @@ description = "AI-powered tech events and CFP tracker for CI/CD, DevOps, and Clo
 readme = "README.md"
 requires-python = ">=3.11"
 dependencies = [
-    "google-genai",
+    "anthropic[vertex]",
     "httpx",
     "beautifulsoup4",
     "fastapi",
@@ -80,7 +80,7 @@ warn_unused_configs = true
 
 [[tool.mypy.overrides]]
 module = [
-    "google.*",
+    "anthropic",
     "bs4",
     "dateutil.*",
     "icalendar",

src/claude_client.py

@@ -0,0 +1,26 @@
+"""Claude on Vertex AI client factory."""
+
+from __future__ import annotations
+
+from anthropic import AsyncAnthropicVertex
+from anthropic.types import Message
+
+from .config import GOOGLE_CLOUD_LOCATION, GOOGLE_CLOUD_PROJECT, is_ai_configured
+from .logging_config import get_logger
+
+logger = get_logger(__name__)
+
+def get_claude_client() -> AsyncAnthropicVertex | None:
+    """Create an async Claude on Vertex AI client, or None when not configured."""
+    if not is_ai_configured():
+        logger.warning("AI search not configured")
+        return None
+    return AsyncAnthropicVertex(
+        project_id=GOOGLE_CLOUD_PROJECT,
+        region=GOOGLE_CLOUD_LOCATION,
+    )
+
+
+def extract_text(response: Message) -> str:
+    """Extract concatenated text blocks from a Claude message response."""
+    return "".join(block.text for block in response.content if block.type == "text")

src/collector/agent.py

@@ -27,6 +27,8 @@ async def collect_all_events(use_ai: bool = True) -> list[Event]:
 
     if use_ai:
         tasks.append(web_search.search_events())
+    else:
+        logger.info("AI search disabled")
 
     results = await asyncio.gather(*tasks, return_exceptions=True)
 

src/collector/sources/web_search.py

@@ -1,18 +1,17 @@
-"""AI-powered web search for event discovery using Gemini."""
+"""AI-powered web search for event discovery using Claude on Vertex AI."""
 
 from __future__ import annotations
 
 import json
 import re
 from datetime import date, datetime
-from typing import Any
+from typing import Any, cast
 
 import httpx
-from google import genai
-from google.genai import types
 
+from ...claude_client import extract_text, get_claude_client
 from ...config import (
-    GEMINI_API_KEY,
+    AI_MODEL,
     REGION_TO_COUNTRIES,
     TARGET_REGIONS,
     TOPICS,
@@ -23,16 +22,23 @@
 
 logger = get_logger(__name__)
 
+WEB_SEARCH_TOOL = [
+    {
+        "type": "web_search_20250305",
+        "name": "web_search",
+        "max_uses": 5,
+    }
+]
+
 
 async def search_events() -> list[Event]:
-    """Use Gemini to search for and extract event information."""
-    if not GEMINI_API_KEY:
-        logger.warning("GEMINI_API_KEY not set, skipping AI search")
+    """Use AI to search for and extract event information."""
+    client = get_claude_client()
+    if client is None:
+        logger.info("Skipping AI web search")
         return []
 
-    logger.info("Starting Gemini AI search")
-
-    client = genai.Client(api_key=GEMINI_API_KEY)
+    logger.info(f"Starting web search with {AI_MODEL} model")
     events = []
 
     for region in TARGET_REGIONS:
@@ -41,11 +47,13 @@ async def search_events() -> list[Event]:
         topics_str = ", ".join(TOPICS[:5])
         current_year = date.today().year
 
-        prompt = f"""Search for upcoming tech conferences and meetups in {region} for {current_year} and {current_year + 1}.
-
-Cover these countries: {countries_str}
+        prompt = f"""Search for upcoming tech conferences and meetups in {region} for {current_year} and {current_year + 1}."""
 
-Focus on events related to: {topics_str}
+        if countries_str:
+            prompt+=f"\n\nCover these countries: {countries_str}"
+        if topics_str:
+            prompt+=f"\n\nFocus on events related to: {topics_str}"
+        prompt+="""
 
 For each event you find, provide the following information in JSON format:
 {{
@@ -66,25 +74,28 @@ async def search_events() -> list[Event]:
   ]
 }}
 
-Only include events that:
-1. Are in one of these countries: {countries_str}
-2. Are related to DevOps, CI/CD, Cloud Native, Kubernetes, or Platform Engineering
-3. Have dates in the future or within the last month
-4. You are reasonably confident about
+Only include events that:"""
+        if {countries_str}:
+            prompt+=f"\n- Are in one of these countries: {countries_str}"
+        if topics_str:
+            prompt+=f"\n- Are related to {topics_str}"
+        prompt+="""
+- Have dates in the future or within the last month
+- You are reasonably confident about
+- Are not already in the events list
 
 Return ONLY the JSON, no other text."""
-
+        print(prompt)
         try:
-            logger.debug("Querying Gemini for region %s", region)
-            response = client.models.generate_content(
-                model="gemini-3-flash-preview",
-                contents=prompt,
-                config=types.GenerateContentConfig(
-                    tools=[types.Tool(google_search=types.GoogleSearch())],
-                ),
+            logger.debug("Querying Claude for region %s", region)
+            response = await client.messages.create(
+                model=AI_MODEL,
+                max_tokens=8192,
+                messages=[{"role": "user", "content": prompt}],
+                tools=cast(Any, WEB_SEARCH_TOOL),
             )
-            content = response.text or ""
-            logger.debug("Gemini response for %s: %d chars", region, len(content))
+            content = extract_text(response)
+            logger.debug("Claude response for %s: %d chars", region, len(content))
             parsed_events = _parse_response(content, region)
             logger.info("Parsed %d events for %s", len(parsed_events), region)
             events.extend(parsed_events)
@@ -97,7 +108,7 @@ async def search_events() -> list[Event]:
 
 
 def _parse_response(content: str, country: str) -> list[Event]:
-    """Parse Gemini's JSON response into Event objects."""
+    """Parse the model JSON response into Event objects."""
     events: list[Event] = []
 
     # Try to extract JSON from the response
@@ -159,8 +170,9 @@ def _parse_response(content: str, country: str) -> list[Event]:
 
 
 async def extract_cfp_details(event_url: str) -> dict:
-    """Use Gemini to extract CFP details from an event website."""
-    if not GEMINI_API_KEY:
+    """Use AI to extract CFP details from an event website."""
+    client = get_claude_client()
+    if client is None:
         return {}
 
     # Fetch the page content
@@ -176,8 +188,6 @@ async def extract_cfp_details(event_url: str) -> dict:
         except Exception:
             return {}
 
-    client = genai.Client(api_key=GEMINI_API_KEY)
-
     prompt = f"""Analyze this event website HTML and extract CFP (Call for Papers/Proposals) information.
 
 HTML content:
@@ -194,14 +204,13 @@ async def extract_cfp_details(event_url: str) -> dict:
 Return ONLY the JSON, no other text."""
 
     try:
-        genai_response = client.models.generate_content(
-            model="gemini-3-flash-preview",
-            contents=prompt,
-            config=types.GenerateContentConfig(
-                tools=[types.Tool(google_search=types.GoogleSearch())],
-            ),
+        genai_response = await client.messages.create(
+            model=AI_MODEL,
+            max_tokens=4096,
+            messages=[{"role": "user", "content": prompt}],
+            tools=cast(Any, WEB_SEARCH_TOOL),
         )
-        content = genai_response.text or ""
+        content = extract_text(genai_response)
         json_match = re.search(r"\{[\s\S]*\}", content)
         if json_match:
             result: dict[str, Any] = json.loads(json_match.group())

src/config.py

@@ -133,9 +133,19 @@ def load_topics(config_file: str | None = None) -> list[str]:
 GLOBAL_CONFERENCES = load_global_conferences()
 TOPICS = load_topics()
 
-GEMINI_API_KEY = os.environ.get("GEMINI_API_KEY", "")
+GOOGLE_CLOUD_PROJECT = os.environ.get(
+    "GOOGLE_CLOUD_PROJECT",
+    os.environ.get("ANTHROPIC_VERTEX_PROJECT_ID", ""),
+)
+GOOGLE_CLOUD_LOCATION = os.environ.get("GOOGLE_CLOUD_LOCATION", "global")
+AI_MODEL = os.environ.get("AI_MODEL", "claude-sonnet-4-6")
 SLACK_WEBHOOK_URL = os.environ.get("SLACK_WEBHOOK_URL", "")
 MEETUP_API_KEY = os.environ.get("MEETUP_API_KEY", "")
 
+
+def is_ai_configured() -> bool:
+    """Return True when Claude on Vertex AI project settings are present."""
+    return bool(GOOGLE_CLOUD_PROJECT)
+
 DATA_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "data")
 EVENTS_FILE = os.path.join(DATA_DIR, "events.json")