Add a run-samples.py script that runs all of the samples (#68)

Maybe eventually this should be "test", but currently they don't do a
very good job testing, since some exit succesfully despite not
working.

Author: msullivan

examples/run-samples.py

@@ -0,0 +1,191 @@
+#!/usr/bin/env python3
+"""Run example samples and report results.
+
+Usage (from repo root):
+    uv run examples/run-samples.py           # text-only samples
+    uv run examples/run-samples.py --image   # also run image samples
+    uv run examples/run-samples.py --video   # also run video samples
+    uv run examples/run-samples.py --all       # run everything
+    uv run examples/run-samples.py --parallel  # run in parallel
+"""
+
+import argparse
+import concurrent.futures
+import dataclasses
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parent.parent
+SAMPLES = REPO / "examples" / "samples"
+
+
+@dataclasses.dataclass
+class Sample:
+    name: str
+    _: dataclasses.KW_ONLY
+    stdin: str | None = None
+
+
+TEXT_SAMPLES = [
+    Sample("stream.py"),
+    Sample("stream_all.py"),
+    Sample("structured_output.py"),
+    Sample("tools_schema.py"),
+    Sample("agent_simple.py"),
+    Sample("agent_custom_loop.py"),
+    Sample("agent_nested.py"),
+    Sample("streaming_tool.py"),
+    Sample("explicit_client.py"),
+    Sample("middleware_simple.py"),
+    Sample("multimodal_input.py"),
+    Sample("check_connection.py"),
+]
+
+IMAGE_SAMPLES = [
+    Sample("image_generation.py"),
+    Sample("image_edit.py"),
+    Sample("inline_image.py"),
+]
+
+VIDEO_SAMPLES = [
+    Sample("video_generation.py"),
+]
+
+# Broken!
+HOOKS_SAMPLES = [
+    Sample("agent_hooks.py", stdin="y\n"),
+    Sample("agent_hooks_serverless.py"),
+]
+
+# Broken!
+MCP_SAMPLES = [
+    Sample("mcp_tools.py"),
+]
+
+
+def _sample_cmd(sample: Sample) -> list[str]:
+    return [
+        "uv",
+        "run",
+        "--frozen",
+        "--with-editable",
+        str(REPO),
+        "python",
+        str(SAMPLES / sample.name),
+    ]
+
+
+_env = {k: v for k, v in os.environ.items() if k != "VIRTUAL_ENV"}
+
+
+def run_sample(sample: Sample) -> bool:
+    print(f"{'=' * 20} {sample.name} {'=' * 20}")
+    sys.stdout.flush()
+    result = subprocess.run(
+        _sample_cmd(sample),
+        env=_env,
+        timeout=120,
+        input=sample.stdin,
+        text=True,
+    )
+    print()
+    sys.stdout.flush()
+    return result.returncode == 0
+
+
+def run_sample_quiet(sample: Sample) -> tuple[str, bool, str]:
+    try:
+        result = subprocess.run(
+            _sample_cmd(sample),
+            env=_env,
+            timeout=120,
+            capture_output=True,
+            text=True,
+            input=sample.stdin,
+        )
+        output = result.stdout + result.stderr
+        return sample.name, result.returncode == 0, output
+    except subprocess.TimeoutExpired:
+        return sample.name, False, "TIMEOUT after 120s"
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Run example samples.")
+    parser.add_argument("--text", action="store_true", help="include text samples")
+    parser.add_argument("--image", action="store_true", help="include image samples")
+    parser.add_argument("--video", action="store_true", help="include video samples")
+    parser.add_argument("--hooks", action="store_true", help="include hook samples")
+    parser.add_argument("--mcp", action="store_true", help="include MCP samples")
+    parser.add_argument("--all", action="store_true", help="run all samples")
+    parser.add_argument(
+        "--parallel", action="store_true", help="run samples in parallel"
+    )
+    args = parser.parse_args()
+
+    has_category = args.text or args.image or args.video or args.hooks or args.mcp
+
+    samples: list[Sample] = []
+    if args.text or args.all or not has_category:
+        samples.extend(TEXT_SAMPLES)
+    if args.image or args.all:
+        samples.extend(IMAGE_SAMPLES)
+    if args.video or args.all:
+        samples.extend(VIDEO_SAMPLES)
+    if args.hooks or args.all:
+        samples.extend(HOOKS_SAMPLES)
+    if args.mcp or args.all:
+        samples.extend(MCP_SAMPLES)
+
+    results: list[tuple[str, bool]] = []
+
+    if args.parallel:
+        outputs: dict[str, str] = {}
+        with concurrent.futures.ThreadPoolExecutor() as pool:
+            futures = {pool.submit(run_sample_quiet, s): s for s in samples}
+            for future in concurrent.futures.as_completed(futures):
+                name, ok, output = future.result()
+                status = "PASS" if ok else "FAIL"
+                print(f"  {status}  {name}")
+                sys.stdout.flush()
+                outputs[name] = output
+                results.append((name, ok))
+
+        passed = sorted(name for name, ok in results if ok)
+        failed = sorted(name for name, ok in results if not ok)
+
+        print()
+        for name in [*passed, *failed]:
+            print(f"{'=' * 20} {name} {'=' * 20}")
+            if outputs[name].strip():
+                print(outputs[name].rstrip())
+            print()
+
+        if failed:
+            sys.exit(1)
+    else:
+        for sample in samples:
+            try:
+                ok = run_sample(sample)
+            except subprocess.TimeoutExpired:
+                print("  TIMEOUT after 120s\n")
+                ok = False
+            results.append((sample.name, ok))
+
+        print("=" * 60)
+        print("Summary:")
+        any_failed = False
+        for name, ok in results:
+            status = "PASS" if ok else "FAIL"
+            print(f"  {status}  {name}")
+            if not ok:
+                any_failed = True
+        print()
+
+        if any_failed:
+            sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()

examples/samples/check_connection.py

@@ -1,49 +1,62 @@
 """Check connection and list models — verify credentials and model availability."""
 
 import asyncio
+import sys
 
 import ai
 
-MODELS = [
-    ai.ai_gateway("anthropic/claude-sonnet-4"),
-    ai.anthropic("claude-sonnet-4-20250514"),
-    ai.openai("gpt-5.4-mini"),
+PROVIDERS: list[tuple[str, ai.Provider, str]] = [
+    ("ai_gateway", ai.ai_gateway, "anthropic/claude-sonnet-4"),
+    ("anthropic", ai.anthropic, "claude-sonnet-4-20250514"),
+    ("openai", ai.openai, "gpt-5.4-mini"),
 ]
 
-PROVIDERS = [
-    ("ai_gateway", ai.ai_gateway),
-    ("anthropic", ai.anthropic),
-    ("openai", ai.openai),
-]
+_failed = False
+
+
+def _fail(msg: str) -> None:
+    global _failed  # noqa: PLW0603
+    _failed = True
+    print(msg)
 
 
-async def _check(model: ai.Model) -> None:
+async def _check(name: str, provider: ai.Provider, model_id: str) -> None:
+    if provider.client().api_key is None:
+        print(f"  [SKIP]  {provider.api_key_env} not set")
+        return
+    model = provider(model_id)
     try:
         ok = await ai.check_connection(model)
-        status = "[OK]  " if ok else "[FAIL]"
+        if ok:
+            print(f"  [OK]    {name}/{model_id}")
+        else:
+            _fail(f"  [FAIL]  {name}/{model_id}")
     except Exception as exc:
-        status = f"[ERR] {exc}"
-    print(f"  {status}  {model.provider}/{model.id}")
+        _fail(f"  [ERR]   {name}/{model_id}: {exc}")
 
 
-async def _list_models(name: str, provider: object) -> None:
+async def _list_models(name: str, provider: ai.Provider) -> None:
+    if provider.client().api_key is None:
+        return
     try:
-        ids: list[str] = await provider.list()  # type: ignore[attr-defined]
-        print(f"  {name}: {len(ids)} models")
-        for mid in ids:
-            print(f"    - {mid}")
+        ids: list[str] = await provider.list()
+        print(f"  {name}: {len(ids)} models (last: {ids[-1]})")
     except Exception as exc:
-        print(f"  {name}: [ERR] {exc}")
+        _fail(f"  {name}: [ERR] {exc}")
 
 
 async def main() -> None:
     print("Checking connections...\n")
-    await asyncio.gather(*[_check(m) for m in MODELS])
+    for name, provider, model_id in PROVIDERS:
+        await _check(name, provider, model_id)
 
     print("\nListing models...\n")
-    await asyncio.gather(*[_list_models(n, p) for n, p in PROVIDERS])
+    for name, provider, _ in PROVIDERS:
+        await _list_models(name, provider)
 
     print()
+    if _failed:
+        sys.exit(1)
 
 
 if __name__ == "__main__":

examples/samples/multimodal_input.py

@@ -7,8 +7,7 @@
 
 model = ai.ai_gateway("anthropic/claude-sonnet-4")
 
-# Load a local image file (replace with your own path).
-image_path = pathlib.Path("sample_image.jpg")
+image_path = pathlib.Path(__file__).parent / "sample_image.jpg"
 image_data = image_path.read_bytes()
 
 messages = [