feat(notion-search): add LLM client abstraction with SDK + CLI fallback

First task of Phase 3b-i (notion semantic search). Adds _search_llm.py
that dispatches to the anthropic SDK when ANTHROPIC_API_KEY is set,
falls back to `claude -p` CLI otherwise, and raises a clear setup
error if neither works. Symlinks _notion_compat.py from 32-notion-writer
so both skills share one source of truth.

3 tests cover the three dispatch paths.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-04-28 07:37:42 +09:00
parent 56f33eca3f
commit 45c68dee61
3 changed files with 144 additions and 0 deletions

View File

@@ -0,0 +1 @@
../../../32-notion-writer/code/scripts/_notion_compat.py

View File

@@ -0,0 +1,67 @@
"""Claude client abstraction. SDK preferred, `claude -p` CLI fallback."""
from __future__ import annotations
import os
import shutil
import subprocess
from typing import Callable
LLMCaller = Callable[..., str] # (prompt, *, model, max_tokens) -> response text
def _have_anthropic_sdk() -> bool:
try:
import anthropic # noqa: F401
return True
except ImportError:
return False
def _have_claude_cli() -> bool:
return shutil.which("claude") is not None
def _call_via_sdk(prompt: str, model: str, max_tokens: int) -> str:
import anthropic
client = anthropic.Anthropic()
response = client.messages.create(
model=model,
max_tokens=max_tokens,
messages=[{"role": "user", "content": prompt}],
)
return response.content[0].text
def _call_via_cli(prompt: str, model: str, max_tokens: int) -> str:
"""Shell out to Claude Code CLI. max_tokens is ignored (CLI handles defaults)."""
result = subprocess.run(
["claude", "-p", prompt, "--model", model],
capture_output=True,
text=True,
check=True,
timeout=60,
)
return result.stdout.strip()
def call_claude(
prompt: str,
*,
model: str = "claude-haiku-4-5",
max_tokens: int = 1000,
) -> str:
"""Send a prompt to Claude.
Tries the anthropic SDK first (requires ANTHROPIC_API_KEY), falls back to
the `claude -p` CLI if available. Raises RuntimeError if neither works.
"""
if _have_anthropic_sdk() and os.getenv("ANTHROPIC_API_KEY"):
return _call_via_sdk(prompt, model, max_tokens)
if _have_claude_cli():
return _call_via_cli(prompt, model, max_tokens)
raise RuntimeError(
"No Claude client available. Either:\n"
" - Install the `anthropic` SDK and set ANTHROPIC_API_KEY, or\n"
" - Install Claude Code CLI (`claude` on PATH)"
)

View File

@@ -0,0 +1,76 @@
#!/usr/bin/env python3
"""Tests for notion_search.py — run with `python3 test_notion_search.py`."""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent))
def _assert(cond, msg):
if not cond:
print(f" ✗ FAIL: {msg}")
raise SystemExit(1)
print(f"{msg}")
def test_call_claude_dispatches_to_sdk_when_available():
"""When SDK is available and ANTHROPIC_API_KEY is set, use SDK path."""
import os
from unittest.mock import patch
import _search_llm
with patch.object(_search_llm, "_have_anthropic_sdk", return_value=True), \
patch.dict(os.environ, {"ANTHROPIC_API_KEY": "sk-fake"}), \
patch.object(_search_llm, "_call_via_sdk", return_value="sdk-response") as sdk_mock, \
patch.object(_search_llm, "_call_via_cli", return_value="cli-response") as cli_mock:
result = _search_llm.call_claude("hello")
_assert(result == "sdk-response", "SDK path returned its response")
_assert(sdk_mock.called, "SDK path was invoked")
_assert(not cli_mock.called, "CLI path was NOT invoked")
def test_call_claude_falls_back_to_cli_when_no_sdk():
"""When SDK is missing but CLI is available, use CLI path."""
from unittest.mock import patch
import _search_llm
with patch.object(_search_llm, "_have_anthropic_sdk", return_value=False), \
patch.object(_search_llm, "_have_claude_cli", return_value=True), \
patch.object(_search_llm, "_call_via_cli", return_value="cli-response") as cli_mock:
result = _search_llm.call_claude("hello")
_assert(result == "cli-response", "CLI path returned its response")
_assert(cli_mock.called, "CLI path was invoked")
def test_call_claude_raises_when_neither_available():
"""Both clients missing → RuntimeError with setup hint."""
from unittest.mock import patch
import _search_llm
with patch.object(_search_llm, "_have_anthropic_sdk", return_value=False), \
patch.object(_search_llm, "_have_claude_cli", return_value=False):
try:
_search_llm.call_claude("hello")
_assert(False, "should have raised RuntimeError")
except RuntimeError as exc:
_assert("ANTHROPIC_API_KEY" in str(exc) or "claude" in str(exc).lower(),
"error mentions setup options")
def run_all():
tests = [
test_call_claude_dispatches_to_sdk_when_available,
test_call_claude_falls_back_to_cli_when_no_sdk,
test_call_claude_raises_when_neither_available,
]
for t in tests:
print(f"\n{t.__name__}")
t()
print("\n" + "=" * 50)
print(f"✅ All {len(tests)} tests passed")
print("=" * 50)
if __name__ == "__main__":
run_all()