A Fun PydanticAI Example For Automating Your Life

Running a PydanticAI CLI app on a timer with github actions cron
Created: 08 February 2025
Last updated: 08 February 2025

Disclaimer: I work on Pydantic Logfire but these opinions are my own.

Introduction

Imagine you want an AI application that periodically fetches some real-world data, analyzes it with an LLM, and saves a neat, validated result without ever worrying about servers. That’s the longing behind this toy Electric Vehicle Charging Growth CLI built on PydanticAI, Typer and scheduled via the GitHub Actions cron feature.

In this post, we’ll walkthrough some of the code (you can see the full source on github), and examine the key areas. The CLI app:

  • Fetches open EV charging station data,
  • Compares it to a metals ETF prices (Platinum, in our demo)
  • Outputs a correlation plus a recommendation.

Is this going to be the trading strategy that changes your life…probably not. Will it make for a fun example - yes.

We’ll also explore why PydanticAI provides a more robust approach than raw LLM SDKs or more complex agent frameworks like LangChain. Finally, we’ll cover potential ways to expand the code, discuss testing strategies, and see how a scheduled GitHub Actions job can run this for free (aside from API call costs).

Code Overview

This project is laid out as a Typer Python CLI app. This is the repo structure:

pydantic-ai-gh-actions-example/
├── .github/
│   └── workflows/
│       └── cron.yml
│
├── mycli/
│   ├── __init__.py
│   └── jobs/
│       └── ev_charging_growth.py
│
├── tests/
│   └── integration/
│       └── test_ev_charging_tools.py
│
├── data.csv
├── settings.py
├── main.py
├── requirements.txt
├── README.md
└── .env  (not in repo - contains secrets)

The main pieces are:

  1. main.py – CLI entry point using typer.
  2. ev_charging_growth.py – Fetches EV charging data, fetches metals ETF data, uses a PydanticAI agent to analyze correlation, and saves outputs.
  3. settings.py – Manages environment variables with pydantic_settings.
  4. tests/ – Integration tests that validate correctness (and gives some ideas for testing PydanticAI).

1. main.py: Your CLI Entry Point

import asyncio
import typer

from mycli.jobs.ev_charging_growth import run_ev_charging_growth

app = typer.Typer(help="GenAI CLI for investment investigations.")

@app.command("ev-charging-growth")
def ev_charging_growth_cli(
    months: int = 5,
    metals_etf: str = "PPLT"
):
    asyncio.run(run_ev_charging_growth(months, metals_etf))

if __name__ == "__main__":
    app()

The code uses Typer to provide a user-friendly CLI. Running:

python -m main ev-charging-growth --months 5 --metals-etf PPLT

kicks off our entire PydanticAI workflow. In other words, this is the function GitHub Actions will call on a schedule. If you didn’t want a CLI, you could invoke run_ev_charging_growth.py from any Python code, but the CLI approach lets us leverage Typer capabilities, such as:

  • Automatic argument parsing based on Python type hints
  • Built-in validation for common types (int, float, Path, etc.)
  • Custom type support through parameter callbacks

2. settings.py: Managing Secrets via Pydantic Settings

from pathlib import Path
from pydantic_settings import BaseSettings

ROOT_DIR = Path(__file__).parent

class Settings(BaseSettings):
    OPENCHARGEMAP_API_KEY: str
    MARKETSTACK_API_KEY: str
    OPENAI_API_KEY: str
    STORAGE_FILE_PATH: Path = ROOT_DIR / "data.csv"
    ...
    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"

settings = Settings()

Nothing AI-specific here, but it’s a great pattern: pydantic-settings to store configuration. You add your environment variables (like OPENAI_API_KEY) in a .env file, and the library automatically reads them. This keeps your secrets out of source control but accessible to GitHub Actions or local dev.

3. ev_charging_growth.py: The Heart of the Operation

Below is the full script that does the real work:

import csv
from datetime import datetime, date, timedelta, timezone
from pathlib import Path
from typing import List, Optional
from collections import defaultdict

import httpx
from pydantic import BaseModel, Field, ConfigDict
from pydantic_ai import Agent, RunContext
from pydantic_ai.models.openai import OpenAIModel
from rich.console import Console
from rich.panel import Panel

from settings import settings

console = Console()


class EvChargingGrowthResult(BaseModel):
    """Result model for EV charging growth analysis."""
    model_config = ConfigDict(frozen=True)

    metals_etf_symbol: str = Field(..., description="Symbol of the analyzed metals ETF")
    correlation: float = Field(..., description="Correlation coefficient between EV growth and metals price")
    recommendation: str = Field(..., description="Analysis-based recommendation")


def already_fetched_today(csv_path: Path, metals_etf: str) -> bool:
    """Check if data for the specified ETF has already been fetched today."""
    today_str = date.today().isoformat()

    if not csv_path.exists():
        return False

    with csv_path.open("r", newline="", encoding="utf-8") as f:
        for row in csv.reader(f):
            # If the date matches today's date and the symbol matches, we've already stored it
            if row[0][:10] == today_str and row[1] == metals_etf:
                return True
    return False


# --------------------------------------
# 1) Create our Agent and register tools
# --------------------------------------
model = OpenAIModel("gpt-4o", api_key=settings.OPENAI_API_KEY)

agent = Agent(
    model,
    deps_type=None,
    result_type=EvChargingGrowthResult,
    system_prompt="""
    You are analyzing EV charging growth rates vs. a metals ETF's monthly prices.
    1. Call the provided tools to fetch real data
    2. Then compute correlation
    3. Return final JSON with (metals_etf_symbol, correlation, recommendation).
    """
)

@agent.tool
async def fetch_ev_charger_growth(
    ctx: RunContext[None],
    months: int
) -> List[float]:
    """Fetch monthly EV charger creation counts from Open Charge Map (real API calls)."""
    base_url = "https://api.openchargemap.io/v3/poi"
    days_ago = months * 30
    since_date = (datetime.now(timezone.utc) - timedelta(days=days_ago)).isoformat()

    async with httpx.AsyncClient() as client:
        response = await client.get(
            base_url,
            params={
                "key": settings.OPENCHARGEMAP_API_KEY,
                "countrycode": "GB",
                "modifiedsince": since_date,
                "maxresults": 5000,
                "compact": "true",
                "verbose": "false"
            },
            timeout=20.0
        )
        response.raise_for_status()
        data = response.json()

    monthly_counts = defaultdict(int)
    now_utc = datetime.now(timezone.utc)
    cutoff_dt = now_utc - timedelta(days=days_ago)

    for poi in data:
        created_str = poi.get("DateCreated")
        if created_str:
            try:
                created_dt = datetime.fromisoformat(created_str.replace("Z", "+00:00"))
                if created_dt >= cutoff_dt:
                    month_key = created_dt.strftime("%Y-%m")
                    monthly_counts[month_key] += 1
            except ValueError:
                continue

    # Create a list of monthly counts in order
    counts_list = []
    for i in reversed(range(months)):
        month_label = (now_utc - timedelta(days=30 * i)).strftime("%Y-%m")
        counts_list.append(float(monthly_counts.get(month_label, 0)))

    return counts_list


@agent.tool
async def fetch_metals_prices(
    ctx: RunContext[None],
    symbol: str,
    months: int
) -> List[float]:
    """Fetch monthly EOD prices for a metals ETF from Marketstack (real API calls)."""
    base_url = "https://api.marketstack.com/v2/eod"
    now_utc = datetime.now(timezone.utc)

    params = {
        "access_key": settings.MARKETSTACK_API_KEY,
        "symbols": symbol,
        "date_from": (now_utc - timedelta(days=months * 30)).strftime("%Y-%m-%d"),
        "date_to": now_utc.strftime("%Y-%m-%d"),
        "limit": 200,
        "sort": "ASC"
    }

    async with httpx.AsyncClient() as client:
        response = await client.get(base_url, params=params, timeout=20.0)
        response.raise_for_status()
        payload = response.json()

    if not (daily_data := payload.get("data")):
        raise ValueError(f"No data found for symbol {symbol}")

    # We want to pick monthly intervals from daily data
    step = max(1, len(daily_data) // months)
    results = [daily_data[i]["close"] for i in range(0, len(daily_data), step)][:months]

    # If we don't have enough months, pad with last known price
    if results and len(results) < months:
        results += [results[-1]] * (months - len(results))

    return results

@agent.result_validator
def finalize_result(ctx: RunContext[None], data: EvChargingGrowthResult) -> EvChargingGrowthResult:
    """Post-process and ensure correlation is in [-1, 1] range."""
    # For safety, clamp correlation
    data.correlation = max(-1.0, min(1.0, data.correlation))
    return data


# ---------------------------------
# 2) Main function to run the agent
# ---------------------------------
async def run_ev_charging_growth(
    months: int = 5,
    metals_etf: str = "PPLT"
) -> Optional[EvChargingGrowthResult]:
    """
    Correlate monthly EV growth rates with a metals ETF price.
    Returns the analysis result if successful, or None if data already fetched today.
    """
    storage_path = Path(settings.STORAGE_FILE_PATH)

    if already_fetched_today(storage_path, metals_etf):
        console.print(
            Panel(f"[yellow]Skipping analysis for {metals_etf} (already fetched today).",
                  title="Analysis Status")
        )
        return None

    console.print(
        Panel(
            f"[blue]Analyzing EV charging growth correlation with {metals_etf} over {months} months",
            title="Analysis Status"
        )
    )

    prompt = f"""
    Correlate EV charger growth vs. {metals_etf} closing prices over {months} months.
    1) Use fetch_ev_charger_growth(months={months})
    2) Use fetch_metals_prices(symbol={metals_etf}, months={months})
    3) Compute correlation
    4) Return a JSON with keys (metals_etf_symbol, correlation, recommendation)
    """
    result = await agent.run(prompt)
    final = result.data

    console.print(
        Panel(
            f"[green]Correlation: {final.correlation:.3f}\n"
            f"Recommendation: {final.recommendation}",
            title=f"Analysis Results for {metals_etf}"
        )
    )

    # Append to CSV
    storage_path.parent.mkdir(parents=True, exist_ok=True)
    with storage_path.open("a", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow([
            datetime.now(timezone.utc).isoformat(),
            final.metals_etf_symbol,
            final.correlation,
            final.recommendation
        ])

    return final

Key Points:

  • We define our expected result type as a Pydantic model: EvChargingGrowthResult, and then pass this to the Agent via the result_type kwarg. This ensures our response is in the format we require. Note that you don’t have to use this, but it’s extremely powerful for making the results of your LLM calls easier to work with, and response errors easier to spot. If you’ve used the Instructor library, it’s similar functionality. Almost all LLM structured output tooling in Python leverages Pydantic under the hood.
  • We define two tools fetch_ev_charger_growth and fetch_metals_prices using the @agent.tool decorator so the LLM can call them.

  • We define a result validator (@agent. result_validator) to constrain the correlation into a [-1,1] min and max range. This an is extremely powerful post-processing step, particularly where you have IO and/or asynchronous results.

  • The final function, run_ev_charging_growth, triggers everything. It calls agent.run(...), and the LLM is free to use the tools as it sees fit in order to yield the typed model result.

There’s a basic persistence & caching function that we don’t need to worry about much (uses the data.csv file) - feel free to swap out for a DB if you need something less of a toy.

4. Testing with pytest

In the tests/integration/test_ev_charging_tools.py module we have this code with Pytest and the pytest-asyncio plugin since our HTTP calls are async:

import pytest
from mycli.jobs.ev_charging_growth import (
    fetch_ev_charger_growth,
    fetch_metals_prices,
)


@pytest.mark.asyncio
async def test_fetch_ev_charger_growth():
    """Test the tool call as a normal sync function."""
    # You can pass a dummy 'ctx' if needed; or None if your code doesn't rely on ctx
    fake_ctx = None  
    months = 5
    growth_rates = await fetch_ev_charger_growth(fake_ctx, months=months)

    assert isinstance(growth_rates, list)
    assert len(growth_rates) == 5
    for rate in growth_rates:
        assert isinstance(rate, float)
        assert 0 <= rate <= 200.0


@pytest.mark.asyncio
async def test_fetch_metals_prices():
    """Test the tool call as a normal sync function."""
    fake_ctx = None
    prices = await fetch_metals_prices(fake_ctx, symbol="PPLT", months=3)

    assert isinstance(prices, list)
    assert len(prices) == 3
    for price in prices:
        assert isinstance(price, float)
        assert 50.0 <= price <= 1000.0

These are real integration tests (no mocks) calling the tool functions. Because we’re doing actual HTTP calls, ensure you have valid API keys in your .env or environment variables. The test verifies that we get lists of floats and not empty data, basically sanity checking the tools.

There are lots of testing options in PydanticAI, including mocking and stub capabilities for the LLM. For instance, you might inject a fake completion that returns a pre-made JSON chunk, verifying the EvChargingGrowthResult parse is correct. Because PydanticAI is built around standard Python function calls, it’s straightforward to:

  • Mock tool calls (like fetch_ev_charger_growth) to avoid real HTTP requests.
  • Stub the agent’s responses.

Why PydanticAI vs. Others

  1. Structured Validation: A raw OpenAI SDK call just returns a string. If the model decides to sing you a haiku about platinum, your code breaks. PydanticAI automatically re-prompts until it meets your schema, or raises an error if it can’t.
  2. Refined Tooling: PydanticAI’s tool calling approach is simple and type-safe (relying on standard Python function signatures, plus docstrings for parameter descriptions).
  3. Less Overhead: PydanticAI is intentionally minimal. You define your agent, add tools, and handle structured results. That’s it. You’re not locked into a monolithic framework. This makes it easier to reason about in a production setting.
  4. Complex Tools IF you need them: PydanticAI has Graph support, but only if you need to do something complex. You can safely ignore it for simpler workflows.

Serverless LLM App Deployment with GitHub Actions

The last piece is scheduling. You can define a .github/workflows/cron.yaml file, for example:

name: "Scheduled Cron Jobs"

on:
  schedule:
    # runs every week at 08:00 UTC
    - cron: "0 8 * * 1"
  workflow_dispatch: {}

jobs:
  run-ev-charging-growth:
    runs-on: ubuntu-latest
    permissions:
      contents: write
    steps:
    - uses: actions/checkout@v4

    - name: Set up Python
      uses: actions/setup-python@v4
      with:
        python-version: "3.11"

    - name: Install uv
      run: |
        curl -LsSf https://astral.sh/uv/install.sh | sh

    - name: Install dependencies
      run: |
        uv sync

    - name: Run EV Charging Growth correlation
      working-directory: $  # Set working directory for this step
      run: |
        uv run python -m main --months 5 --metals-etf PPLT
    
    - name: Commit and push
      run: |
        # Configure Git with the GitHub Actions bot identity
        git config --global user.name "github-actions"
        git config --global user.email "41898282+github-actions[bot]@users.noreply.github.com"
        
        # If data.csv was modified, let's commit it:
        git add data.csv
        git diff --cached --quiet && echo "No changes to commit" || (
          git commit -m "chore: update data.csv"
          git push
        )

    env:
        OPENCHARGEMAP_API_KEY: ${{ secrets.OPENCHARGEMAP_API_KEY }}
        MARKETSTACK_API_KEY: ${{ secrets.MARKETSTACK_API_KEY }}
        OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}

This approach is effectively free for public repos (thanks to included minutes ), and even on private repos a daily job uses only minimal quota.

The best part is zero ops overhead. Deployment is just pushing your code to the repo. The next scheduled run picks up your changes. No server to maintain, and logs are available in the Actions UI if something fails. Secrets (API keys, etc.) are stored in GitHub and injected as env vars, so you don’t expose creds in code. Each run is in a clean environment, ensuring consistent execution. Low hassle.

You could also use any other CI tool you prefer.

How You Could Develop The App Further

1) Monitoring Pretty soon, you’ll want some easy observability for your LLM app - I’m biased, but I’d recommend Pydantic Logfire.

2) Notifications and Email Handling
You could extend run_ev_charging_growth to automatically send a Slack or Gmail notification if the correlation is above a certain threshold. PydanticAI’s dependency injection system would let you pass a Gmail client, for instance, into your tool function (@agent.tool), and your agent can call it at runtime.

3) PydanticAI Graphs
For more complex workflows—maybe comparing multiple ETFs, or branching logic based on correlation, PydanticAI provides a Graph feature (currently in beta). You can define nodes as states, letting the LLM decide the next step. It’s powerful for multi-step automation while still retaining structured validation each step of the way.

Conclusion

By combining PydanticAI’s structured approach with a lightweight CLI and GitHub Actions scheduling, you get a robust, low-ops AI workflow. Rather than dealing with arbitrary text responses, you define typed models that guarantee consistent data every run. Rather than spinning up a custom server or cron, you rely on GitHub’s ephemeral runners.

For an “AI Ops” pattern that’s simple to manage yet powerful, it’s hard to beat. Whether you adapt this EV charging example or swap in a different data source, the underlying approach remains the same: a short CLI, robust validation, free cron, and unstoppable AI automation.

Happy (assisted) coding!

Best of the Blog

Category

Tags