diff --git a/.agents/skills/batchor-dev/SKILL.md b/.agents/skills/batchor-dev/SKILL.md index 9d90cbb..c1f15a6 100644 --- a/.agents/skills/batchor-dev/SKILL.md +++ b/.agents/skills/batchor-dev/SKILL.md @@ -14,13 +14,14 @@ Use this skill for normal contributor work in this repository. 3. Load only the design docs that match the task: - `docs/design_docs/ARCHITECTURE.md` for package layout and runtime boundaries - `docs/design_docs/OPENAI_BATCHING.md` for provider-specific batching behavior + - `docs/design_docs/ANTHROPIC_BATCHING.md` for Anthropic Message Batches behavior - `docs/design_docs/STORAGE_AND_RUNS.md` for durable state, run lifecycle, and retention - `docs/smoke-test.md` for the full validation matrix ## Repo map - `src/batchor/runtime/`: orchestration, run handles, validation, retry, token budgeting -- `src/batchor/providers/`: provider interfaces and OpenAI Batch implementation +- `src/batchor/providers/`: provider interfaces and OpenAI, Anthropic, and Gemini implementations - `src/batchor/storage/`: SQLite, Postgres, and in-memory durability backends - `src/batchor/sources/`: file-backed checkpointable item sources - `src/batchor/artifacts/`: durable request/output artifact handling diff --git a/README.md b/README.md index 52a14ab..cf25017 100644 --- a/README.md +++ b/README.md @@ -12,7 +12,7 @@ - library-first run controls - a small operator CLI for CSV and JSONL jobs -It is intentionally narrow today: OpenAI is the CLI default, Gemini is opt-in through `batchor[gemini]`, SQLite is the CLI durability backend, and the Python API exposes the broadest configuration surface. +It is intentionally narrow today: OpenAI is the CLI default, Anthropic and Gemini are opt-in through provider extras, SQLite is the CLI durability backend, and the Python API exposes the broadest configuration surface. ## What problem it solves @@ -37,6 +37,7 @@ Most OpenAI Batch examples stop at "upload a JSONL file and poll until it finish Built-in implementations: - `OpenAIProviderConfig` + `OpenAIBatchProvider` +- `AnthropicProviderConfig` + `AnthropicBatchProvider` for Claude Message Batches - `GeminiProviderConfig` + `GeminiBatchProvider` for text-only Gemini Batch jobs - `SQLiteStorage` - `PostgresStorage` as an opt-in durable control-plane backend @@ -53,7 +54,7 @@ Important constraints: - the CLI supports file-backed inputs only - users still own selecting and ordering input files or partitions - the built-in CLI uses SQLite durability only -- the CLI supports OpenAI plus Gemini Developer API and Vertex AI text jobs +- the CLI supports OpenAI, Anthropic, Gemini Developer API, and Vertex AI text jobs - Gemini support is text-only for now and does not build multimodal requests - structured-output rehydration requires an importable module-level Pydantic model - raw output artifacts are retained by default and must be exported before raw pruning @@ -102,6 +103,7 @@ graph LR subgraph providers["providers/"] OpenAI["OpenAIBatchProvider"] + Anthropic["AnthropicBatchProvider"] Gemini["GeminiBatchProvider"] end @@ -123,6 +125,7 @@ graph LR BatchRunner --> Run BatchRunner --> Executor BatchRunner --> OpenAI + BatchRunner --> Anthropic BatchRunner --> Gemini BatchRunner --> SQLite BatchRunner --> LocalFS @@ -165,6 +168,12 @@ For Gemini Batch support, install the optional extra: pip install "batchor[gemini]" ``` +For Anthropic Message Batches support: + +```bash +pip install "batchor[anthropic]" +``` + ## Agent setup Batchor keeps contributor tooling separate from the tools intended for researchers and downstream projects: @@ -195,8 +204,8 @@ Supported Python versions: For Python API usage, auth resolution is: -1. explicit provider config credentials such as `OpenAIProviderConfig(api_key=...)` or `GeminiProviderConfig(api_key=...)` -2. ambient provider environment variables, currently `OPENAI_API_KEY` or `GEMINI_API_KEY` +1. explicit provider config credentials such as `OpenAIProviderConfig(api_key=...)`, `AnthropicProviderConfig(api_key=...)`, or `GeminiProviderConfig(api_key=...)` +2. ambient provider environment variables: `OPENAI_API_KEY`, `ANTHROPIC_API_KEY`, or `GEMINI_API_KEY` 3. Vertex AI Application Default Credentials when `GeminiProviderConfig(vertexai=True, ...)` or `GOOGLE_GENAI_USE_VERTEXAI=true` is used The Python library does not auto-load `.env`. @@ -226,6 +235,29 @@ run = runner.run_and_wait( print(run.results()[0].output_text) ``` +### Anthropic text job + +```python +from batchor import AnthropicProviderConfig, BatchItem, BatchJob, BatchRunner, PromptParts + + +runner = BatchRunner(storage="memory") +run = runner.run_and_wait( + BatchJob( + items=[BatchItem(item_id="row1", payload="Summarize this text")], + build_prompt=lambda item: PromptParts(prompt=item.payload), + provider_config=AnthropicProviderConfig( + model="claude-sonnet-4-5", + max_tokens=1024, + ), + ) +) + +print(run.results()[0].output_text) +``` + +Anthropic support maps Batchor attempt identifiers deterministically to Claude-safe `custom_id` values and supports system prompts, `message_params`, and structured outputs through `output_config.format`. + ### Gemini text job ```python diff --git a/docs/design_docs/ANTHROPIC_BATCHING.md b/docs/design_docs/ANTHROPIC_BATCHING.md new file mode 100644 index 0000000..0b0d955 --- /dev/null +++ b/docs/design_docs/ANTHROPIC_BATCHING.md @@ -0,0 +1,33 @@ +# Anthropic Batching + +The built-in Anthropic provider targets the Claude Message Batches API. + +## Request mapping + +Each Batchor item attempt becomes one Anthropic batch request: + +- Batchor's durable correlation identifier is deterministically hashed into Anthropic's 64-character-safe `custom_id` alphabet +- `PromptParts.prompt` becomes one user message +- `PromptParts.system_prompt` becomes the top-level `system` parameter +- `model` and `max_tokens` come from `AnthropicProviderConfig` +- `message_params` supplies optional Messages API parameters +- structured output schemas use `output_config.format` + +Batchor rejects provider-owned fields in `message_params` so replayed requests cannot silently change model, token limit, messages, system prompt, or streaming behavior. + +## Submission and lifecycle + +Anthropic accepts request objects directly rather than an uploaded JSONL file. The provider therefore stages Batchor's durable request artifact in memory only for the upload/create boundary, then calls `messages.batches.create(requests=...)`. + +Anthropic `in_progress` maps to Batchor's active status and `ended` maps to `completed`. Once ended, the provider streams `messages.batches.results(...)` into Batchor's raw output artifact. Result order is not assumed; correlation always uses `custom_id`. + +The four Anthropic result types map as follows: + +- `succeeded`: Batchor success +- `errored`, `canceled`, and `expired`: Batchor item error, subject to the configured retry policy + +## Limits and retention + +Anthropic currently limits a Message Batch to 100,000 requests or 256 MB. Batchor's default chunk limits are lower than both limits. Claude batch results remain available from Anthropic for 29 days; Batchor downloads and retains them according to its artifact policy, so durable runs do not depend on that remote retention window after ingestion. + +Message Batches are not eligible for Anthropic Zero Data Retention. Users should account for Anthropic's batch retention policy when selecting workloads. diff --git a/docs/design_docs/ARCHITECTURE.md b/docs/design_docs/ARCHITECTURE.md index a47b9a1..d932912 100644 --- a/docs/design_docs/ARCHITECTURE.md +++ b/docs/design_docs/ARCHITECTURE.md @@ -61,6 +61,7 @@ graph TB subgraph providers["providers/"] BatchProvider["BatchProvider (ABC)"] OpenAIProvider["OpenAIBatchProvider"] + AnthropicProvider["AnthropicBatchProvider"] GeminiProvider["GeminiBatchProvider"] ProviderRegistry end @@ -95,6 +96,7 @@ graph TB MemoryStateStore -.->|implements| StateStore BatchRunner -->|submits/polls| BatchProvider OpenAIProvider -.->|implements| BatchProvider + AnthropicProvider -.->|implements| BatchProvider GeminiProvider -.->|implements| BatchProvider BatchRunner -->|stores artifacts| ArtifactStore LocalArtifactStore -.->|implements| ArtifactStore @@ -126,7 +128,7 @@ one logical source, while callers remain responsible for selecting and ordering the child sources up front. Provider adaptation is intentionally concentrated behind `BatchProvider`. -The runtime stores one durable internal custom identifier per item attempt, while each provider maps that identifier to its own request shape. OpenAI uses `custom_id`; the Gemini Developer API uses `key`; Vertex AI uses a request label that is returned with the original request in GCS output. +The runtime stores one durable internal custom identifier per item attempt, while each provider maps that identifier to its own request shape. OpenAI and Anthropic use `custom_id`; the Gemini Developer API uses `key`; Vertex AI uses a request label that is returned with the original request in GCS output. Provider hooks also own response-text extraction so structured-output parsing can stay generic across provider payload shapes. ## Main user-facing flow @@ -311,6 +313,7 @@ Owns provider-facing abstractions and implementations: - base provider contract - provider registry - OpenAI Batch implementation +- Anthropic Message Batches implementation The provider layer is responsible for: @@ -436,7 +439,7 @@ That split gives `batchor`: ## Current invariants 1. Public execution is run-oriented: `start()`, `get_run()`, `run_and_wait()`. -2. OpenAI Batch and text-only Gemini Batch are built-in Python and CLI providers; OpenAI remains the CLI default. +2. OpenAI Batch, Anthropic Message Batches, and text-only Gemini Batch are built-in Python and CLI providers; OpenAI remains the CLI default. 3. SQLite is the default durable backend. 4. Postgres is an opt-in durable backend for shared control-plane state. 5. Structured outputs require a module-level Pydantic v2 model for rehydration. diff --git a/docs/design_docs/OPENAI_BATCHING.md b/docs/design_docs/OPENAI_BATCHING.md index 719fa49..2d338d3 100644 --- a/docs/design_docs/OPENAI_BATCHING.md +++ b/docs/design_docs/OPENAI_BATCHING.md @@ -2,11 +2,11 @@ This document describes the OpenAI-specific behavior inside `batchor`. -OpenAI remains the default and most feature-complete provider path. Gemini has its own design note in [`GEMINI_BATCHING.md`](GEMINI_BATCHING.md), while this page focuses only on OpenAI Batch semantics. +OpenAI remains the default and most feature-complete provider path. Anthropic and Gemini have their own design notes in [`ANTHROPIC_BATCHING.md`](ANTHROPIC_BATCHING.md) and [`GEMINI_BATCHING.md`](GEMINI_BATCHING.md), while this page focuses only on OpenAI Batch semantics. ## Current behavior -The Python API and CLI both support OpenAI. The CLI is OpenAI-only today. +The Python API and CLI both support OpenAI, which remains the CLI default. ## Request construction diff --git a/docs/design_docs/ROADMAP.md b/docs/design_docs/ROADMAP.md index 989e35b..8973e3a 100644 --- a/docs/design_docs/ROADMAP.md +++ b/docs/design_docs/ROADMAP.md @@ -4,17 +4,13 @@ This file tracks important work that should be explicit in the extracted reposit ## Next Practical Steps -- add Postgres storage backend -- extend resumable ingestion beyond the built-in CSV/JSONL sources - add automated retention windows on top of the explicit export/prune lifecycle -- add more input adapters beyond CSV and JSONL -- expose non-OpenAI providers through CLI workflows once provider-specific auth and flags are stable - add CLI structured-output workflows if the importability story can stay durable and predictable ## Longer-Term Ideas -- additional provider coverage beyond OpenAI and text-only Gemini -- artifact store abstraction +- additional provider coverage beyond OpenAI, Anthropic, and text-only Gemini +- remote/shared artifact store implementations beyond the local filesystem abstraction - partial-result streaming APIs - richer CLI or operator workflow beyond the current file-backed text-job surface diff --git a/docs/doc-map.md b/docs/doc-map.md index 6159496..f76141b 100644 --- a/docs/doc-map.md +++ b/docs/doc-map.md @@ -19,6 +19,7 @@ This page explains what each document is for so readers do not have to guess whi | `design_docs/BOUNDARY_AND_PHILOSOPHY.md` | Ownership boundary between `batchor`, storage/artifacts, and user pipelines. | | `design_docs/ARCHITECTURE.md` | Canonical runtime diagrams, package structure, main flows, and extension seams. | | `design_docs/OPENAI_BATCHING.md` | OpenAI request construction, token budgeting, splitting, and batch polling behavior. | +| `design_docs/ANTHROPIC_BATCHING.md` | Anthropic Message Batches request construction, polling, result parsing, limits, and retention. | | `design_docs/GEMINI_BATCHING.md` | Gemini text-only request construction, batch polling, response parsing, and current limits. | | `design_docs/STORAGE_AND_RUNS.md` | Durable `Run` lifecycle, rehydration, checkpoints, control state, artifact retention, and operator semantics. | | `design_docs/STORAGE_MIGRATIONS.md` | SQLite schema-versioning and migration guidance. | @@ -40,3 +41,4 @@ This page explains what each document is for so readers do not have to guess whi - Incremental terminal-result reads/exports are documented in the Python API and storage docs. - Raw output/error artifact retention is now configurable per run through `ArtifactPolicy`. - Gemini text-only Batch support is available through the Python API and default provider registry. +- Anthropic Message Batches support is available through the Python API, CLI, and default provider registry. diff --git a/docs/getting-started/cli.md b/docs/getting-started/cli.md index 2f78191..79acd7f 100644 --- a/docs/getting-started/cli.md +++ b/docs/getting-started/cli.md @@ -40,6 +40,13 @@ pip install "batchor[gemini]" echo "GEMINI_API_KEY=..." > .env ``` +For Anthropic: + +```bash +pip install "batchor[anthropic]" +echo "ANTHROPIC_API_KEY=..." > .env +``` + ## Start a run from JSONL ```bash @@ -66,6 +73,20 @@ Use `--prompt-template` when the prompt should be derived from several fields. Exactly one of `--prompt-field` or `--prompt-template` is required. +## Start an Anthropic run + +```bash +batchor start \ + --input input/items.jsonl \ + --id-field id \ + --prompt-field text \ + --provider anthropic \ + --model claude-sonnet-4-5 \ + --anthropic-max-tokens 1024 +``` + +Use `--anthropic-message-params` for an additional JSON object such as `'{"temperature":0.2}'`. + ## Start a Gemini Developer API run ```bash diff --git a/docs/getting-started/installation.md b/docs/getting-started/installation.md index 8b71340..a3a25bd 100644 --- a/docs/getting-started/installation.md +++ b/docs/getting-started/installation.md @@ -22,6 +22,12 @@ pip install "batchor[gemini]" Gemini support is text-only and is available through both the Python API and CLI. +Anthropic support uses the optional Anthropic SDK dependency: + +```bash +pip install "batchor[anthropic]" +``` + ## What gets installed The package includes: @@ -29,6 +35,7 @@ The package includes: - the Python library - the `batchor` CLI - the built-in OpenAI provider integration +- the built-in Anthropic provider integration when `batchor[anthropic]` is installed - the built-in Gemini provider integration when `batchor[gemini]` is installed - SQLite and Postgres storage implementations @@ -63,11 +70,16 @@ For Gemini Developer API usage, authentication resolution is: 1. `GeminiProviderConfig(api_key=...)` 2. `GEMINI_API_KEY` +For Anthropic usage, authentication resolution is: + +1. `AnthropicProviderConfig(api_key=...)` +2. `ANTHROPIC_API_KEY` + Vertex AI uses Application Default Credentials and resolves project/location from explicit `GeminiProviderConfig` fields or `GOOGLE_CLOUD_PROJECT` and `GOOGLE_CLOUD_LOCATION`. Vertex batches also require a writable `gs://` staging prefix. The Python library does not auto-load `.env`. -The CLI loads a local `.env` as a convenience for operator usage. It resolves `OPENAI_API_KEY` for OpenAI, `GEMINI_API_KEY` for the Gemini Developer API, and the documented Google Cloud environment variables for Vertex AI. Rehydrated `status`, `wait`, and result operations load the same `.env` before reconstructing a provider. +The CLI loads a local `.env` as a convenience for operator usage. It resolves `OPENAI_API_KEY`, `ANTHROPIC_API_KEY`, `GEMINI_API_KEY`, and the documented Google Cloud environment variables for Vertex AI. Rehydrated `status`, `wait`, and result operations load the same `.env` before reconstructing a provider. ## Storage defaults diff --git a/docs/getting-started/python-api.md b/docs/getting-started/python-api.md index b748873..2410ca9 100644 --- a/docs/getting-started/python-api.md +++ b/docs/getting-started/python-api.md @@ -39,6 +39,31 @@ print(run.results()[0].output_text) Use `storage="memory"` only for tests or short-lived local experiments. For durable runs, use the default SQLite storage or an explicit backend. +## Anthropic text job + +Install `batchor[anthropic]`, then use the same runner lifecycle with an Anthropic provider config: + +```python +from batchor import AnthropicProviderConfig, BatchItem, BatchJob, BatchRunner, PromptParts + + +runner = BatchRunner(storage="memory") +run = runner.run_and_wait( + BatchJob( + items=[BatchItem(item_id="row1", payload="Summarize this text")], + build_prompt=lambda item: PromptParts(prompt=item.payload), + provider_config=AnthropicProviderConfig( + model="claude-sonnet-4-5", + max_tokens=1024, + ), + ) +) + +print(run.results()[0].output_text) +``` + +If `api_key` is omitted, the provider resolves `ANTHROPIC_API_KEY`. Extra Messages API parameters such as `temperature` can be supplied through `message_params`; Batchor reserves the fields it constructs itself. + ## Gemini text job Gemini Batch support is available through the Python API after installing `batchor[gemini]`. @@ -131,6 +156,7 @@ Notes: - `output` is the parsed Pydantic object - `output_text` preserves the raw text that was parsed - with `GeminiProviderConfig`, the same `structured_output=` argument is sent through Gemini `generation_config.response_json_schema` +- with `AnthropicProviderConfig`, it is sent through Anthropic `output_config.format` ## Durable run lifecycle diff --git a/docs/index.md b/docs/index.md index 46ecbbe..bf77d68 100644 --- a/docs/index.md +++ b/docs/index.md @@ -56,7 +56,7 @@ If that is the mental model you were missing from the generated docs, start with ## Current surface -- Built-in providers: OpenAI Batch, plus text-only Gemini Batch for Python API usage +- Built-in providers: OpenAI Batch, Anthropic Message Batches, and text-only Gemini Batch through Python and CLI workflows - Durable storage: SQLite by default, Postgres as an opt-in control-plane backend - Ephemeral storage: in-memory state store - Artifact backend: local filesystem via `LocalArtifactStore` diff --git a/docs/policies/contributing.md b/docs/policies/contributing.md index 8d4e3e3..8e33758 100644 --- a/docs/policies/contributing.md +++ b/docs/policies/contributing.md @@ -13,7 +13,7 @@ Before submitting changes: Please keep changes aligned with the documented package boundary: - durable batch execution -- OpenAI Batch provider support +- OpenAI, Anthropic, and Gemini Batch provider support - SQLite-backed local durability by default Out-of-scope proposals are still useful, but should usually start as an issue discussing fit before implementation. diff --git a/docs/reference/api.md b/docs/reference/api.md index 6280a7a..12cc128 100644 --- a/docs/reference/api.md +++ b/docs/reference/api.md @@ -20,6 +20,7 @@ Most users only need a small subset of the package: - `BatchRunner`: start, resume, and run orchestration - `Run`: refresh, wait, inspect, export, and prune - `OpenAIProviderConfig`: built-in provider config +- `AnthropicProviderConfig`: built-in Anthropic Message Batches config - `GeminiProviderConfig`: built-in Gemini provider config for text batch jobs; the common fields are also exposed by the CLI - `SQLiteStorage` and `PostgresStorage`: durable control-plane backends - `CompositeItemSource`, `CsvItemSource`, `JsonlItemSource`, and `ParquetItemSource`: deterministic item streaming @@ -73,6 +74,15 @@ This is the built-in Gemini Batch implementation. Most consumers only need `Gemi show_root_heading: true heading_level: 2 +## Anthropic provider + +Install `batchor[anthropic]` for the SDK dependency. Most consumers only need `AnthropicProviderConfig`. + +::: batchor.providers.anthropic + options: + show_root_heading: true + heading_level: 2 + ## Sources These sources support durable resume through source fingerprints and checkpoints. diff --git a/docs/smoke-test.md b/docs/smoke-test.md index 03d1e4a..cdd5d50 100644 --- a/docs/smoke-test.md +++ b/docs/smoke-test.md @@ -8,6 +8,7 @@ This guide defines the minimum validation bar for `batchor`. - verify durable run handling and SQLite persistence still work - verify artifact-store wiring still supports replay, export, and prune - verify OpenAI-specific batching logic through fake-provider integration tests +- verify Anthropic Message Batches wiring through fake-client and opt-in live tests - verify Gemini provider wiring through fake-client integration tests - verify the documentation site still builds cleanly in strict mode @@ -79,6 +80,7 @@ uv run pytest tests/unit/test_batchor_tokens.py tests/unit/test_batchor_sqlite_s uv run pytest tests/unit/test_batchor_artifacts.py tests/unit/test_batchor_storage_contracts.py --no-cov -q uv run pytest tests/integration/test_batchor_runner.py --no-cov -q uv run pytest tests/unit/test_batchor_gemini_provider.py tests/integration/test_batchor_gemini_runner.py --no-cov -q +uv run pytest tests/unit/test_batchor_anthropic_provider.py --no-cov -q ``` Expected: @@ -107,6 +109,7 @@ Expected: - completed submitted items report consumed attempts consistently across storage backends - OpenAI request splitting and enqueue-limit logic still behave as expected - Gemini text-only request construction, batch polling normalization, response parsing, and structured-output validation still behave as expected +- Anthropic request construction, safe correlation IDs, polling normalization, result parsing, and structured-output validation still behave as expected - wait-mode refresh cycles keep draining immediately after poll/submission progress, without introducing idle poll sleeps while more local work can be sent - structured-output parsing remains stable @@ -188,7 +191,18 @@ The root `.env` or shell must also provide `GEMINI_API_KEY` with available Devel Cost controls: -- three total items across both live tests +- three total items across the Gemini live transports - text output only - manual/local only - not part of default CI or release automation + +## Live Anthropic smoke + +Manual only. This submits one minimal text request through the normal SQLite-backed runtime: + +```bash +export BATCHOR_RUN_LIVE_ANTHROPIC=1 +uv run --extra anthropic pytest tests/integration/test_batchor_live_anthropic.py --no-cov -q +``` + +The root `.env` or shell must provide `ANTHROPIC_API_KEY`. The test defaults to `claude-haiku-4-5`, 64 output tokens, and a 15-minute timeout. Override these with `BATCHOR_LIVE_ANTHROPIC_MODEL` and `BATCHOR_LIVE_ANTHROPIC_TIMEOUT_SEC`. diff --git a/mkdocs.yml b/mkdocs.yml index a7dca85..aff304e 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -89,6 +89,7 @@ nav: - Boundary & Philosophy: design_docs/BOUNDARY_AND_PHILOSOPHY.md - OpenAI Batching: design_docs/OPENAI_BATCHING.md - Gemini Batching: design_docs/GEMINI_BATCHING.md + - Anthropic Batching: design_docs/ANTHROPIC_BATCHING.md - Storage & Runs: design_docs/STORAGE_AND_RUNS.md - Storage Migrations: design_docs/STORAGE_MIGRATIONS.md - Roadmap: design_docs/ROADMAP.md diff --git a/plugins/batchor-agent-tools/scripts/batchor_repo_mcp.py b/plugins/batchor-agent-tools/scripts/batchor_repo_mcp.py index 8b5fb0f..2ab2b41 100644 --- a/plugins/batchor-agent-tools/scripts/batchor_repo_mcp.py +++ b/plugins/batchor-agent-tools/scripts/batchor_repo_mcp.py @@ -20,7 +20,7 @@ def _repo_path(relative_path: str) -> str: def _project_guide() -> str: lines = [ - "batchor is a durable OpenAI and Gemini Batch runner with typed Pydantic outputs, resumable runs, durable artifacts, and a narrow operator CLI.", + "batchor is a durable OpenAI, Anthropic, and Gemini Batch runner with typed Pydantic outputs, resumable runs, durable artifacts, and a narrow operator CLI.", "", f"Repo root: {REPO_ROOT}", "Read these first:", @@ -28,6 +28,7 @@ def _project_guide() -> str: f"- README.md: {_repo_path('README.md')}", f"- Architecture: {_repo_path('docs/design_docs/ARCHITECTURE.md')}", f"- OpenAI batching: {_repo_path('docs/design_docs/OPENAI_BATCHING.md')}", + f"- Anthropic batching: {_repo_path('docs/design_docs/ANTHROPIC_BATCHING.md')}", f"- Gemini batching: {_repo_path('docs/design_docs/GEMINI_BATCHING.md')}", f"- Storage and runs: {_repo_path('docs/design_docs/STORAGE_AND_RUNS.md')}", f"- Smoke tests: {_repo_path('docs/smoke-test.md')}", @@ -61,11 +62,11 @@ def _project_guide() -> str: "- uv run pytest tests/integration/test_batchor_runner.py --no-cov -q", ], "provider": [ - "Use when changing OpenAI or Gemini request shaping, batching, file upload/download handling, or provider normalization.", + "Use when changing OpenAI, Anthropic, or Gemini request shaping, batching, file handling, or provider normalization.", "Commands:", "- uv run ty check src", "- uv run pytest -q", - "- uv run pytest tests/unit/test_batchor_openai_provider.py tests/unit/test_batchor_gemini_provider.py tests/unit/test_batchor_tokens.py --no-cov -q", + "- uv run pytest tests/unit/test_batchor_openai_provider.py tests/unit/test_batchor_anthropic_provider.py tests/unit/test_batchor_gemini_provider.py tests/unit/test_batchor_tokens.py --no-cov -q", ], "storage": [ "Use when changing SQLite/Postgres state handling, lifecycle persistence, or storage contracts.", @@ -94,6 +95,11 @@ def _project_guide() -> str: ("README", "README.md", "Public scope, quickstart, and mental model."), ("Architecture", "docs/design_docs/ARCHITECTURE.md", "Package layout and runtime boundaries."), ("OpenAI batching", "docs/design_docs/OPENAI_BATCHING.md", "Provider request construction and enqueue policy."), + ( + "Anthropic batching", + "docs/design_docs/ANTHROPIC_BATCHING.md", + "Message Batches request and result behavior.", + ), ("Gemini batching", "docs/design_docs/GEMINI_BATCHING.md", "Developer API and Vertex AI transports."), ( "Storage and runs", @@ -118,6 +124,11 @@ def _project_guide() -> str: "docs/design_docs/GEMINI_BATCHING.md", "Developer inline/File modes, Vertex GCS transport, and provider limits.", ), + ( + "Anthropic batching", + "docs/design_docs/ANTHROPIC_BATCHING.md", + "Message Batches request shaping, result streaming, and provider limits.", + ), ("Architecture", "docs/design_docs/ARCHITECTURE.md", "Provider layer boundary."), ("Smoke tests", "docs/smoke-test.md", "Targeted provider validation."), ], diff --git a/plugins/batchor/scripts/batchor_user_mcp.py b/plugins/batchor/scripts/batchor_user_mcp.py index d31574b..6f56f41 100755 --- a/plugins/batchor/scripts/batchor_user_mcp.py +++ b/plugins/batchor/scripts/batchor_user_mcp.py @@ -28,7 +28,7 @@ def list_tools() -> list[dict[str, Any]]: "shared_workers": {"type": "boolean"}, "provider": { "type": "string", - "enum": ["openai", "gemini", "vertex"], + "enum": ["openai", "anthropic", "gemini", "vertex"], }, }, "required": ["input_kind", "structured_output", "shared_workers", "provider"], @@ -42,7 +42,7 @@ def list_tools() -> list[dict[str, Any]]: "type": "object", "properties": { "surface": {"type": "string", "enum": ["cli", "python"]}, - "provider": {"type": "string", "enum": ["openai", "gemini", "vertex"]}, + "provider": {"type": "string", "enum": ["openai", "anthropic", "gemini", "vertex"]}, "model": {"type": "string"}, "id_field": {"type": "string"}, "prompt_field": {"type": "string"}, @@ -70,11 +70,11 @@ def _choose_workflow(args: dict[str, Any]) -> str: provider = str(args["provider"]) use_cli = input_kind in {"csv", "jsonl"} and not structured and not shared surface = "CLI" if use_cli else "Python API" - install = ( - 'python -m pip install "batchor[gemini]"' - if provider in {"gemini", "vertex"} - else "python -m pip install batchor" - ) + install = { + "anthropic": 'python -m pip install "batchor[anthropic]"', + "gemini": 'python -m pip install "batchor[gemini]"', + "vertex": 'python -m pip install "batchor[gemini]"', + }.get(provider, "python -m pip install batchor") storage = "Postgres plus an artifact root shared by all workers" if shared else "default SQLite durability" source = { "csv": "CsvItemSource", @@ -102,7 +102,12 @@ def _starter(args: dict[str, Any]) -> str: prompt_field = str(args["prompt_field"]) if surface == "cli": backend = "vertex" if provider == "vertex" else "developer" - provider_flag = "" if provider == "openai" else f" --provider gemini --gemini-backend {backend}" + if provider == "anthropic": + provider_flag = " --provider anthropic --anthropic-max-tokens 1024" + elif provider in {"gemini", "vertex"}: + provider_flag = f" --provider gemini --gemini-backend {backend}" + else: + provider_flag = "" return ( "Inspect `batchor start --help` for the installed version, then adapt this without running it:\n\n" "batchor start \\\n" @@ -111,10 +116,18 @@ def _starter(args: dict[str, Any]) -> str: f" --prompt-field {prompt_field} \\\n" f" --model {model}{provider_flag}\n" ) - is_gemini = provider in {"gemini", "vertex"} - config = "GeminiProviderConfig" if is_gemini else "OpenAIProviderConfig" - install = '"batchor[gemini]"' if is_gemini else "batchor" - provider_args = f'model="{model}", vertexai=True' if provider == "vertex" else f'model="{model}"' + if provider == "anthropic": + config = "AnthropicProviderConfig" + install = '"batchor[anthropic]"' + provider_args = f'model="{model}", max_tokens=1024' + elif provider in {"gemini", "vertex"}: + config = "GeminiProviderConfig" + install = '"batchor[gemini]"' + provider_args = f'model="{model}", vertexai=True' if provider == "vertex" else f'model="{model}"' + else: + config = "OpenAIProviderConfig" + install = "batchor" + provider_args = f'model="{model}"' return f'''Install {install} with the project's dependency manager, then adapt this module:\n\nfrom batchor import BatchItem, BatchJob, BatchRunner, {config}, PromptParts\n\n\ndef start_job(records: list[dict[str, str]], run_id: str):\n runner = BatchRunner()\n job = BatchJob(\n items=[\n BatchItem(item_id=row["{id_field}"], payload=row)\n for row in records\n ],\n build_prompt=lambda item: PromptParts(prompt=item.payload["{prompt_field}"]),\n provider_config={config}({provider_args}),\n )\n return runner.start(job, run_id=run_id)\n\nDo not call this against real records until the user approves the upload and cost.\n''' diff --git a/plugins/batchor/skills/use-batchor/SKILL.md b/plugins/batchor/skills/use-batchor/SKILL.md index 7ec4e85..9a1ee23 100644 --- a/plugins/batchor/skills/use-batchor/SKILL.md +++ b/plugins/batchor/skills/use-batchor/SKILL.md @@ -1,6 +1,6 @@ --- name: use-batchor -description: Turn datasets, prompts, or existing Python data pipelines into durable OpenAI or Gemini Batch workflows with Batchor. Use when a researcher or downstream project needs to process CSV, JSONL, Parquet, or application records with resumable LLM batches, typed outputs, result export, or run operations; also use to diagnose or improve an existing Batchor integration. Do not use for contributing to the Batchor library itself. +description: Turn datasets, prompts, or existing Python data pipelines into durable OpenAI, Anthropic, or Gemini Batch workflows with Batchor. Use when a researcher or downstream project needs to process CSV, JSONL, Parquet, or application records with resumable LLM batches, typed outputs, result export, or run operations; also use to diagnose or improve an existing Batchor integration. Do not use for contributing to the Batchor library itself. --- # Use Batchor @@ -11,7 +11,7 @@ Build the smallest reliable Batchor workflow that fits the user's data and opera 1. Inspect the project and a small sample of the input schema when available. 2. Identify or safely infer: - - provider: OpenAI, Gemini Developer API, or Vertex AI + - provider: OpenAI, Anthropic, Gemini Developer API, or Vertex AI - input: CSV, JSONL, Parquet, or application records - stable item identifier and prompt fields - plain text versus structured Pydantic output @@ -31,7 +31,7 @@ If the Batchor MCP tools are available, call `batchor_choose_workflow` before im ## Implement safely -1. Add `batchor` to the project's normal dependency manager. Use `batchor[gemini]` only for Gemini. +1. Add `batchor` to the project's normal dependency manager. Use `batchor[anthropic]` for Anthropic or `batchor[gemini]` for Gemini. 2. Keep credentials in the project's existing secret mechanism. Python callers do not get automatic `.env` loading from Batchor. 3. Preserve a stable `run_id` outside the process so a fresh process can call `get_run()` or resume `start(..., run_id=...)`. 4. Prefer `CsvItemSource`, `JsonlItemSource`, `ParquetItemSource`, or another checkpointed source when restart-safe ingestion matters. diff --git a/plugins/batchor/skills/use-batchor/references/cli-workflows.md b/plugins/batchor/skills/use-batchor/references/cli-workflows.md index 553ff6f..29cee76 100644 --- a/plugins/batchor/skills/use-batchor/references/cli-workflows.md +++ b/plugins/batchor/skills/use-batchor/references/cli-workflows.md @@ -11,6 +11,8 @@ export OPENAI_API_KEY="..." Use `python -m pip install "batchor[gemini]"` and the relevant Gemini or Google Cloud credentials for Gemini. +For Anthropic, install `batchor[anthropic]`, set `ANTHROPIC_API_KEY`, and select `--provider anthropic`. Anthropic also requires `--anthropic-max-tokens`. + ## Start a run For JSONL containing `id` and `text`: diff --git a/plugins/batchor/skills/use-batchor/references/python-pipelines.md b/plugins/batchor/skills/use-batchor/references/python-pipelines.md index 82f05e5..7f7da10 100644 --- a/plugins/batchor/skills/use-batchor/references/python-pipelines.md +++ b/plugins/batchor/skills/use-batchor/references/python-pipelines.md @@ -47,6 +47,8 @@ class Finding(BaseModel): Pass it as `structured_output=Finding` on `BatchJob`. Consume validated `result.output`, while handling failed items explicitly. For OpenAI strict schemas, object fields must satisfy the provider's required/closed-object rules. +For Anthropic, install `batchor[anthropic]` and use `AnthropicProviderConfig(model=..., max_tokens=...)`. Extra Messages parameters belong in `message_params`; Batchor maps structured output through `output_config.format`. + ## Durable file sources Prefer `CsvItemSource`, `JsonlItemSource`, or `ParquetItemSource` for large files. Give every row a stable `item_id`. Re-enter `start(job, run_id=the_same_id)` only with the same logical source and compatible job configuration. diff --git a/pyproject.toml b/pyproject.toml index 34f2447..1e4c272 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "batchor" -version = "0.0.4" +version = "0.0.5" description = "Structured-first provider Batch runner with typed Pydantic results." readme = "README.md" license = { file = "LICENSE" } @@ -19,7 +19,7 @@ dependencies = [ "typer>=0.16.0,<1", "python-dotenv>=1.0.1,<2", ] -keywords = ["openai", "gemini", "batch", "sqlite", "cli", "pydantic"] +keywords = ["openai", "anthropic", "gemini", "batch", "sqlite", "cli", "pydantic"] classifiers = [ "Development Status :: 3 - Alpha", "Environment :: Console", @@ -42,6 +42,9 @@ Issues = "https://github.com/AnsonDev42/batchor/issues" batchor = "batchor.cli:app" [project.optional-dependencies] +anthropic = [ + "anthropic>=0.71.0,<1", +] gemini = [ "google-genai>=1.55.0,<2", "google-cloud-storage>=3.4.0,<4", diff --git a/src/batchor/__init__.py b/src/batchor/__init__.py index 726a567..f759f94 100644 --- a/src/batchor/__init__.py +++ b/src/batchor/__init__.py @@ -39,6 +39,7 @@ StructuredOutputSchemaError, ) from batchor.core.models import ( + AnthropicProviderConfig, ArtifactExportResult, ArtifactPolicy, ArtifactPruneResult, @@ -61,6 +62,7 @@ TerminalResultsPage, TextItemResult, ) +from batchor.providers.anthropic import AnthropicBatchProvider from batchor.providers.base import ( BatchProvider, ProviderConfig, @@ -83,6 +85,8 @@ from batchor.storage.state import MemoryStateStore, StateStore __all__ = [ + "AnthropicBatchProvider", + "AnthropicProviderConfig", "ArtifactStore", "ArtifactPolicy", "BatchProvider", diff --git a/src/batchor/cli.py b/src/batchor/cli.py index 6e94e95..64358e6 100644 --- a/src/batchor/cli.py +++ b/src/batchor/cli.py @@ -27,6 +27,7 @@ from batchor.core.enums import GeminiBatchInputMode, OpenAIEndpoint, ProviderKind from batchor.core.models import ( + AnthropicProviderConfig, ArtifactExportResult, ArtifactPruneResult, BatchJob, @@ -185,7 +186,19 @@ def _provider_config( google_cloud_project: str, google_cloud_location: str, gemini_generation_config: str | None, + anthropic_max_tokens: int, + anthropic_message_params: str | None, ) -> ProviderConfig: + if provider is ProviderKind.ANTHROPIC: + return AnthropicProviderConfig( + model=model, + max_tokens=anthropic_max_tokens, + poll_interval_sec=poll_interval_sec, + message_params=_json_object_option( + anthropic_message_params, + option_name="--anthropic-message-params", + ), + ) if provider is ProviderKind.GEMINI: return GeminiProviderConfig( model=model, @@ -369,6 +382,12 @@ def start_command( "--gemini-generation-config", help="JSON object merged into each Gemini generation request.", ), + anthropic_max_tokens: int = typer.Option(1024, "--anthropic-max-tokens"), + anthropic_message_params: str | None = typer.Option( + None, + "--anthropic-message-params", + help="JSON object merged into each Anthropic Messages request.", + ), ) -> None: storage: SQLiteStorage | None = None try: @@ -398,6 +417,8 @@ def start_command( google_cloud_project=google_cloud_project, google_cloud_location=google_cloud_location, gemini_generation_config=gemini_generation_config, + anthropic_max_tokens=anthropic_max_tokens, + anthropic_message_params=anthropic_message_params, ) runner, storage = _runner_for_path( db_path=db_path, diff --git a/src/batchor/core/enums.py b/src/batchor/core/enums.py index 1e775fd..ab9e986 100644 --- a/src/batchor/core/enums.py +++ b/src/batchor/core/enums.py @@ -73,6 +73,7 @@ class ProviderKind(StrEnum): OPENAI = "openai" GEMINI = "gemini" + ANTHROPIC = "anthropic" class GeminiBatchInputMode(StrEnum): diff --git a/src/batchor/core/models.py b/src/batchor/core/models.py index 0362773..e670914 100644 --- a/src/batchor/core/models.py +++ b/src/batchor/core/models.py @@ -371,6 +371,73 @@ def from_payload(cls, payload: JSONObject) -> GeminiProviderConfig: ) +@dataclass(frozen=True) +class AnthropicProviderConfig(ProviderConfig): + """Configuration for the built-in Anthropic Message Batches provider.""" + + model: str + max_tokens: int + api_key: str = "" + poll_interval_sec: float = 1.0 + message_params: JSONObject = field(default_factory=dict) + + def __post_init__(self) -> None: + if not self.model.strip(): + raise ValueError("model must be a non-empty string") + if self.max_tokens < 1: + raise ValueError("max_tokens must be >= 1") + if self.poll_interval_sec <= 0: + raise ValueError("poll_interval_sec must be > 0") + params = _json_object_copy(self.message_params, label="message_params") + reserved = {"model", "max_tokens", "messages", "system", "stream"} & params.keys() + if reserved: + raise ValueError(f"message_params contains reserved fields: {', '.join(sorted(reserved))}") + object.__setattr__(self, "message_params", params) + + @property + def provider_kind(self) -> ProviderKind: + return ProviderKind.ANTHROPIC + + def to_payload(self) -> JSONObject: + return { + "model": self.model, + "max_tokens": self.max_tokens, + "api_key": self.api_key, + "poll_interval_sec": self.poll_interval_sec, + "message_params": dict(self.message_params), + } + + def to_public_payload(self) -> JSONObject: + payload = self.to_payload() + payload.pop("api_key", None) + return payload + + @classmethod + def from_payload(cls, payload: JSONObject) -> AnthropicProviderConfig: + model = payload.get("model") + max_tokens = payload.get("max_tokens") + api_key = payload.get("api_key", "") + poll_interval_sec = payload.get("poll_interval_sec", 1.0) + message_params = payload.get("message_params", {}) + if not isinstance(model, str): + raise TypeError("model must be a string") + if not isinstance(max_tokens, int): + raise TypeError("max_tokens must be an int") + if not isinstance(api_key, str): + raise TypeError("api_key must be a string") + if not isinstance(poll_interval_sec, int | float): + raise TypeError("poll_interval_sec must be numeric") + if not isinstance(message_params, dict): + raise TypeError("message_params must be a JSON object") + return cls( + model=model, + max_tokens=max_tokens, + api_key=api_key, + poll_interval_sec=float(poll_interval_sec), + message_params=cast(JSONObject, message_params), + ) + + @dataclass(frozen=True) class OpenAIProviderConfig(ProviderConfig): """Configuration for the built-in OpenAI Batch provider. diff --git a/src/batchor/providers/anthropic.py b/src/batchor/providers/anthropic.py new file mode 100644 index 0000000..ccabe97 --- /dev/null +++ b/src/batchor/providers/anthropic.py @@ -0,0 +1,177 @@ +"""Anthropic Message Batches provider implementation.""" + +from __future__ import annotations + +import hashlib +import json +import os +from importlib import import_module +from pathlib import Path +from typing import Any, cast +from uuid import uuid4 + +from batchor.core.models import AnthropicProviderConfig, PromptParts +from batchor.core.types import BatchRemoteRecord, BatchRequestLine, JSONObject +from batchor.providers.base import BatchProvider, StructuredOutputSchema +from batchor.runtime.tokens import estimate_request_tokens + +_INPUT_PREFIX = "anthropic-input://" +_RESULTS_PREFIX = "anthropic-results://" + + +def _anthropic_custom_id(custom_id: str) -> str: + """Map Batchor identifiers to Anthropic's 64-character safe alphabet.""" + return f"b{hashlib.sha256(custom_id.encode('utf-8')).hexdigest()[:40]}" + + +def resolve_anthropic_api_key(config: AnthropicProviderConfig) -> str: + api_key = config.api_key or os.getenv("ANTHROPIC_API_KEY", "") + if not api_key: + raise ValueError("Anthropic API key is required; set ANTHROPIC_API_KEY or pass api_key") + return api_key + + +class AnthropicBatchProvider(BatchProvider): + """Adapter for Anthropic's asynchronous Message Batches API.""" + + def __init__(self, config: AnthropicProviderConfig, *, client: Any | None = None) -> None: + self.config = config + if client is None: + anthropic = import_module("anthropic") + client = anthropic.Anthropic(api_key=resolve_anthropic_api_key(config)) + self.client = client + self._inputs: dict[str, list[JSONObject]] = {} + self._results: dict[str, str] = {} + + def build_request_line( + self, + *, + custom_id: str, + prompt_parts: PromptParts, + structured_output: StructuredOutputSchema | None = None, + ) -> BatchRequestLine: + params: JSONObject = dict(self.config.message_params) + params.update( + { + "model": self.config.model, + "max_tokens": self.config.max_tokens, + "messages": [{"role": "user", "content": prompt_parts.prompt}], + } + ) + if prompt_parts.system_prompt: + params["system"] = prompt_parts.system_prompt + if structured_output is not None: + params["output_config"] = {"format": {"type": "json_schema", "schema": structured_output.schema}} + return {"custom_id": _anthropic_custom_id(custom_id), "body": params} + + def with_request_correlation_id(self, request_line: BatchRequestLine, custom_id: str) -> BatchRequestLine: + updated = dict(request_line) + updated["custom_id"] = _anthropic_custom_id(custom_id) + return cast(BatchRequestLine, updated) + + def upload_input_file(self, input_path: str | Path) -> str: + identifier = f"{_INPUT_PREFIX}{uuid4().hex}" + self._inputs[identifier] = self._parse_jsonl(Path(input_path).read_text(encoding="utf-8")) + return identifier + + def delete_input_file(self, file_id: str) -> None: + self._inputs.pop(file_id, None) + + def create_batch(self, *, input_file_id: str, metadata: dict[str, str] | None = None) -> BatchRemoteRecord: + del metadata + try: + rows = self._inputs.pop(input_file_id) + except KeyError as exc: + raise ValueError(f"unknown Anthropic staged input: {input_file_id}") from exc + requests = [] + for row in rows: + custom_id = row.get("custom_id") + params = row.get("body") + if not isinstance(custom_id, str) or not isinstance(params, dict): + raise ValueError("Anthropic batch rows require custom_id and body fields") + requests.append({"custom_id": custom_id, "params": params}) + return self._normalize(self.client.messages.batches.create(requests=requests)) + + def get_batch(self, batch_id: str) -> BatchRemoteRecord: + remote = self.client.messages.batches.retrieve(batch_id) + normalized = self._normalize(remote) + if normalized["status"] == "completed": + result_id = f"{_RESULTS_PREFIX}{batch_id}" + results = self.client.messages.batches.results(batch_id) + self._results[result_id] = "".join( + json.dumps(self._object_to_dict(result), ensure_ascii=False) + "\n" for result in results + ) + normalized["output_file_id"] = result_id + return normalized + + def download_file_content(self, file_id: str) -> str: + try: + return self._results[file_id] + except KeyError as exc: + raise ValueError(f"unknown Anthropic results identifier: {file_id}") from exc + + def parse_batch_output( + self, *, output_content: str | None, error_content: str | None + ) -> tuple[dict[str, JSONObject], dict[str, JSONObject], list[JSONObject]]: + raw = self._parse_jsonl(output_content or "") + self._parse_jsonl(error_content or "") + successes: dict[str, JSONObject] = {} + errors: dict[str, JSONObject] = {} + for record in raw: + custom_id = record.get("custom_id") + result = record.get("result") + if not isinstance(custom_id, str) or not isinstance(result, dict): + continue + (successes if result.get("type") == "succeeded" else errors)[custom_id] = record + return successes, errors, raw + + def extract_response_text(self, response_record: JSONObject) -> str: + result = response_record.get("result") + message = result.get("message") if isinstance(result, dict) else None + content = message.get("content") if isinstance(message, dict) else None + if not isinstance(content, list): + return "" + return "\n".join( + cast(str, block["text"]) + for block in content + if isinstance(block, dict) and block.get("type") == "text" and isinstance(block.get("text"), str) + ) + + def estimate_request_tokens(self, request_line: BatchRequestLine, *, chars_per_token: int) -> int: + return estimate_request_tokens(request_line, chars_per_token=chars_per_token, model=self.config.model) + + @staticmethod + def _parse_jsonl(content: str) -> list[JSONObject]: + records: list[JSONObject] = [] + for raw_line in content.splitlines(): + if not raw_line.strip(): + continue + value = json.loads(raw_line) + if not isinstance(value, dict): + raise ValueError("batch jsonl records must be JSON objects") + records.append(cast(JSONObject, value)) + return records + + @classmethod + def _normalize(cls, obj: Any) -> BatchRemoteRecord: + payload = cls._object_to_dict(obj) + batch_id = payload.get("id") + processing_status = payload.get("processing_status") + if not isinstance(batch_id, str) or not isinstance(processing_status, str): + raise TypeError("Anthropic batch response is missing id or processing_status") + counts = payload.get("request_counts") + return BatchRemoteRecord( + id=batch_id, + status="completed" if processing_status == "ended" else "in_progress", + output_file_id=None, + error_file_id=None, + request_counts=cast(Any, counts if isinstance(counts, dict) else {}), + errors=None, + ) + + @staticmethod + def _object_to_dict(obj: Any) -> JSONObject: + if isinstance(obj, dict): + return cast(JSONObject, obj) + if hasattr(obj, "model_dump"): + return cast(JSONObject, obj.model_dump(mode="json")) + raise TypeError(f"cannot normalize object {type(obj)!r}") diff --git a/src/batchor/providers/registry.py b/src/batchor/providers/registry.py index 4d9aef7..2496f79 100644 --- a/src/batchor/providers/registry.py +++ b/src/batchor/providers/registry.py @@ -16,7 +16,7 @@ from batchor.providers.base import BatchProvider, ProviderConfig if TYPE_CHECKING: - from batchor.core.models import GeminiProviderConfig, OpenAIProviderConfig + from batchor.core.models import AnthropicProviderConfig, GeminiProviderConfig, OpenAIProviderConfig type ProviderFactory = Callable[[ProviderConfig], BatchProvider] type ProviderConfigLoader = Callable[[JSONObject], ProviderConfig] @@ -133,7 +133,8 @@ def build_default_provider_registry() -> ProviderRegistry: Returns: A :class:`ProviderRegistry` with OpenAI and Gemini registered. """ - from batchor.core.models import GeminiProviderConfig, OpenAIProviderConfig + from batchor.core.models import AnthropicProviderConfig, GeminiProviderConfig, OpenAIProviderConfig + from batchor.providers.anthropic import AnthropicBatchProvider from batchor.providers.gemini import GeminiBatchProvider from batchor.providers.openai import OpenAIBatchProvider @@ -148,6 +149,11 @@ def build_default_provider_registry() -> ProviderRegistry: factory=lambda config: GeminiBatchProvider(_require_gemini_config(config)), loader=GeminiProviderConfig.from_payload, ) + registry.register( + kind=ProviderKind.ANTHROPIC, + factory=lambda config: AnthropicBatchProvider(_require_anthropic_config(config)), + loader=AnthropicProviderConfig.from_payload, + ) return registry @@ -165,3 +171,11 @@ def _require_gemini_config(config: ProviderConfig) -> GeminiProviderConfig: if not isinstance(config, GeminiProviderConfig): raise TypeError(f"expected {ProviderKind.GEMINI.value} config, got {type(config).__name__}") return config + + +def _require_anthropic_config(config: ProviderConfig) -> AnthropicProviderConfig: + from batchor.core.models import AnthropicProviderConfig + + if not isinstance(config, AnthropicProviderConfig): + raise TypeError(f"expected {ProviderKind.ANTHROPIC.value} config, got {type(config).__name__}") + return config diff --git a/tests/integration/test_batchor_live_anthropic.py b/tests/integration/test_batchor_live_anthropic.py new file mode 100644 index 0000000..b237a98 --- /dev/null +++ b/tests/integration/test_batchor_live_anthropic.py @@ -0,0 +1,60 @@ +"""Opt-in live smoke coverage for Anthropic Message Batches.""" + +from __future__ import annotations + +import os +from pathlib import Path +from uuid import uuid4 + +import pytest +from dotenv import find_dotenv, load_dotenv + +from batchor import AnthropicProviderConfig, BatchItem, BatchJob, BatchRunner, PromptParts, SQLiteStorage + +load_dotenv(find_dotenv(usecwd=True), override=False) + +pytestmark = [pytest.mark.integration, pytest.mark.live] + + +@pytest.mark.skipif( + os.getenv("BATCHOR_RUN_LIVE_ANTHROPIC") != "1", + reason="set BATCHOR_RUN_LIVE_ANTHROPIC=1 to run Anthropic smoke coverage", +) +def test_live_anthropic_text_job_smoke(tmp_path: Path) -> None: + """Complete one real Message Batch through the durable runtime.""" + api_key = os.getenv("ANTHROPIC_API_KEY", "") + if not api_key: + pytest.skip("ANTHROPIC_API_KEY is required for live Anthropic smoke coverage") + + storage = SQLiteStorage(path=tmp_path / "anthropic-live.sqlite3") + try: + runner = BatchRunner(storage=storage, temp_root=tmp_path / "artifacts") + run = runner.start( + BatchJob( + items=[BatchItem(item_id="anthropic-smoke-1", payload="Reply with exactly: batchor-anthropic-live-ok")], + build_prompt=lambda item: PromptParts(prompt=item.payload), + provider_config=AnthropicProviderConfig( + api_key=api_key, + model=os.getenv("BATCHOR_LIVE_ANTHROPIC_MODEL", "claude-haiku-4-5"), + max_tokens=64, + poll_interval_sec=5.0, + message_params={"temperature": 0}, + ), + ), + run_id=f"live_anthropic_{uuid4().hex[:10]}", + ) + + run.wait( + timeout=float(os.getenv("BATCHOR_LIVE_ANTHROPIC_TIMEOUT_SEC", "900")), + poll_interval=5.0, + ) + result = run.results()[0] + normalized = (result.output_text or "").strip().strip("`\"'").rstrip(".!?").lower() + + assert result.error is None + assert normalized == "batchor-anthropic-live-ok" + inventory = storage.get_artifact_inventory(run_id=run.run_id) + assert len(inventory.request_artifact_paths) == 1 + assert len(inventory.output_artifact_paths) == 1 + finally: + storage.close() diff --git a/tests/unit/test_agent_tooling.py b/tests/unit/test_agent_tooling.py index 575668a..4258c25 100644 --- a/tests/unit/test_agent_tooling.py +++ b/tests/unit/test_agent_tooling.py @@ -55,6 +55,7 @@ def test_project_guide_points_to_repo_docs(): assert "AGENTS.md" in text assert "docs/design_docs/ARCHITECTURE.md" in text + assert "docs/design_docs/ANTHROPIC_BATCHING.md" in text assert "uv run pytest -q" in text @@ -161,3 +162,45 @@ def test_user_starter_supports_vertex_python_config(): assert "GeminiProviderConfig" in text assert "vertexai=True" in text assert 'Install "batchor[gemini]"' in text + + +def test_user_tools_support_anthropic_workflows(): + module = _load_module( + "plugins/batchor/scripts/batchor_user_mcp.py", + "batchor_user_mcp", + ) + + workflow = module.call_tool( + "batchor_choose_workflow", + { + "input_kind": "jsonl", + "structured_output": False, + "shared_workers": False, + "provider": "anthropic", + }, + )["content"][0]["text"] + cli = module.call_tool( + "batchor_starter", + { + "surface": "cli", + "provider": "anthropic", + "model": "claude-haiku-4-5", + "id_field": "record_id", + "prompt_field": "abstract", + }, + )["content"][0]["text"] + python = module.call_tool( + "batchor_starter", + { + "surface": "python", + "provider": "anthropic", + "model": "claude-haiku-4-5", + "id_field": "record_id", + "prompt_field": "abstract", + }, + )["content"][0]["text"] + + assert 'pip install "batchor[anthropic]"' in workflow + assert "--provider anthropic --anthropic-max-tokens 1024" in cli + assert "AnthropicProviderConfig" in python + assert "max_tokens=1024" in python diff --git a/tests/unit/test_batchor_anthropic_provider.py b/tests/unit/test_batchor_anthropic_provider.py new file mode 100644 index 0000000..7123055 --- /dev/null +++ b/tests/unit/test_batchor_anthropic_provider.py @@ -0,0 +1,117 @@ +from __future__ import annotations + +from pathlib import Path + +import pytest +from pydantic import BaseModel + +from batchor.core.models import AnthropicProviderConfig, PromptParts +from batchor.providers.anthropic import AnthropicBatchProvider, resolve_anthropic_api_key +from batchor.providers.base import StructuredOutputSchema +from batchor.runtime.validation import model_output_schema + + +class _ClassificationResult(BaseModel): + label: str + + +class _Batches: + def __init__(self) -> None: + self.requests: list[dict[str, object]] = [] + + def create(self, *, requests): # noqa: ANN001 + self.requests = requests + return {"id": "msgbatch_123", "processing_status": "in_progress", "request_counts": {}} + + def retrieve(self, batch_id: str): # noqa: ANN201 + assert batch_id == "msgbatch_123" + return {"id": batch_id, "processing_status": "ended", "request_counts": {"succeeded": 1}} + + def results(self, batch_id: str): # noqa: ANN201 + assert batch_id == "msgbatch_123" + return iter( + [ + { + "custom_id": "b525575f3ea077094fb3f5f3de2fbab850d5b5db5", + "result": { + "type": "succeeded", + "message": {"content": [{"type": "text", "text": "done"}]}, + }, + }, + {"custom_id": "b3dfa78c04a9678358e0ba53f3cf4b9723be1df2a", "result": {"type": "expired"}}, + ] + ) + + +class _Messages: + def __init__(self) -> None: + self.batches = _Batches() + + +class _Client: + def __init__(self) -> None: + self.messages = _Messages() + + +def test_build_request_line_supports_system_and_structured_output() -> None: + _, schema = model_output_schema(_ClassificationResult) + provider = AnthropicBatchProvider( + AnthropicProviderConfig( + api_key="k", + model="claude-sonnet-4-5", + max_tokens=512, + message_params={"temperature": 0.2}, + ), + client=_Client(), + ) + + line = provider.build_request_line( + custom_id="row1:a1", + prompt_parts=PromptParts(prompt="hello", system_prompt="classify"), + structured_output=StructuredOutputSchema("classification_result", schema), + ) + + assert line["custom_id"] == "b525575f3ea077094fb3f5f3de2fbab850d5b5db5" + assert line["body"]["messages"] == [{"role": "user", "content": "hello"}] + assert line["body"]["system"] == "classify" + assert line["body"]["temperature"] == 0.2 + assert line["body"]["output_config"]["format"]["schema"] == schema + replayed = provider.with_request_correlation_id(line, "row2:a1") + assert replayed["custom_id"] == "b3dfa78c04a9678358e0ba53f3cf4b9723be1df2a" + + +def test_submit_poll_download_and_parse(tmp_path: Path) -> None: + client = _Client() + provider = AnthropicBatchProvider( + AnthropicProviderConfig(api_key="k", model="claude-sonnet-4-5", max_tokens=512), + client=client, + ) + input_path = tmp_path / "requests.jsonl" + input_path.write_text( + '{"custom_id":"b525575f3ea077094fb3f5f3de2fbab850d5b5db5",' + '"body":{"model":"claude-sonnet-4-5","max_tokens":512,' + '"messages":[{"role":"user","content":"hello"}]}}\n', + encoding="utf-8", + ) + + input_id = provider.upload_input_file(input_path) + created = provider.create_batch(input_file_id=input_id) + remote = provider.get_batch(created["id"]) + output = provider.download_file_content(remote["output_file_id"]) + successes, errors, raw = provider.parse_batch_output(output_content=output, error_content=None) + + assert client.messages.batches.requests[0]["custom_id"] == "b525575f3ea077094fb3f5f3de2fbab850d5b5db5" + assert created["status"] == "in_progress" + assert remote["status"] == "completed" + assert set(successes) == {"b525575f3ea077094fb3f5f3de2fbab850d5b5db5"} + assert set(errors) == {"b3dfa78c04a9678358e0ba53f3cf4b9723be1df2a"} + assert provider.extract_response_text(successes["b525575f3ea077094fb3f5f3de2fbab850d5b5db5"]) == "done" + assert len(raw) == 2 + + +def test_config_validation_and_api_key_resolution(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("ANTHROPIC_API_KEY", "env-key") + config = AnthropicProviderConfig(model="claude-sonnet-4-5", max_tokens=1) + assert resolve_anthropic_api_key(config) == "env-key" + with pytest.raises(ValueError, match="reserved fields"): + AnthropicProviderConfig(model="claude-sonnet-4-5", max_tokens=1, message_params={"stream": True}) diff --git a/tests/unit/test_batchor_architecture.py b/tests/unit/test_batchor_architecture.py index f3dd571..5362f03 100644 --- a/tests/unit/test_batchor_architecture.py +++ b/tests/unit/test_batchor_architecture.py @@ -4,6 +4,7 @@ from pathlib import Path from batchor import ( + AnthropicProviderConfig, GeminiBatchProvider, GeminiProviderConfig, MemoryStateStore, @@ -88,6 +89,31 @@ def test_default_provider_registry_can_dump_secretless_gemini_config() -> None: assert "api_key" not in payload["config"] +def test_default_provider_registry_round_trips_anthropic_config() -> None: + registry = build_default_provider_registry() + config = AnthropicProviderConfig( + api_key="k", + model="claude-sonnet-4-5", + max_tokens=1024, + message_params={"temperature": 0.1}, + ) + + payload = registry.dump_config(config) + loaded = registry.load_config(payload) + assert loaded.provider_kind is ProviderKind.ANTHROPIC + assert loaded == config + + +def test_default_provider_registry_can_dump_secretless_anthropic_config() -> None: + registry = build_default_provider_registry() + config = AnthropicProviderConfig(api_key="secret", model="claude-sonnet-4-5", max_tokens=1024) + + payload = registry.dump_config(config, include_secrets=False) + + assert payload["provider_kind"] == ProviderKind.ANTHROPIC.value + assert "api_key" not in payload["config"] + + def test_storage_registry_supports_explicit_backend_factories(tmp_path: Path) -> None: sqlite_path = tmp_path / "registry.sqlite3" provider_registry = build_default_provider_registry() diff --git a/tests/unit/test_batchor_cli.py b/tests/unit/test_batchor_cli.py index 258664c..5eed3fd 100644 --- a/tests/unit/test_batchor_cli.py +++ b/tests/unit/test_batchor_cli.py @@ -9,7 +9,7 @@ from batchor.cli import create_app from batchor.core.enums import GeminiBatchInputMode -from batchor.core.models import GeminiProviderConfig, OpenAIProviderConfig, PromptParts +from batchor.core.models import AnthropicProviderConfig, GeminiProviderConfig, OpenAIProviderConfig, PromptParts from batchor.providers.openai import OpenAIBatchProvider @@ -199,6 +199,42 @@ def provider_factory(config): # noqa: ANN001 assert config.generation_config == {"temperature": 0.2} +def test_cli_start_builds_anthropic_config(tmp_path: Path) -> None: + input_path = tmp_path / "items.jsonl" + input_path.write_text('{"id":"row1","text":"hello"}\n', encoding="utf-8") + provider = _FakeCliProvider() + seen_configs: list[object] = [] + + result = CliRunner().invoke( + create_app(provider_factory=lambda config: (seen_configs.append(config), provider)[1]), + [ + "start", + "--input", + str(input_path), + "--id-field", + "id", + "--prompt-field", + "text", + "--provider", + "anthropic", + "--model", + "claude-sonnet-4-5", + "--anthropic-max-tokens", + "2048", + "--anthropic-message-params", + '{"temperature":0.2}', + "--db-path", + str(tmp_path / "cli.sqlite3"), + ], + ) + + assert result.exit_code == 0 + config = seen_configs[0] + assert isinstance(config, AnthropicProviderConfig) + assert config.max_tokens == 2048 + assert config.message_params == {"temperature": 0.2} + + def test_cli_start_builds_gemini_vertex_config(tmp_path: Path) -> None: input_path = tmp_path / "items.jsonl" input_path.write_text('{"id":"row1","text":"hello"}\n', encoding="utf-8") diff --git a/uv.lock b/uv.lock index 0b23af5..aa02fb7 100644 --- a/uv.lock +++ b/uv.lock @@ -24,6 +24,25 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/78/b6/6307fbef88d9b5ee7421e68d78a9f162e0da4900bc5f5793f6d3d0e34fb8/annotated_types-0.7.0-py3-none-any.whl", hash = "sha256:1f02e8b43a8fbbc3f3e0d4f0f4bfc8131bcb4eebe8849b8e5c773f3a1c582a53", size = 13643, upload-time = "2024-05-20T21:33:24.1Z" }, ] +[[package]] +name = "anthropic" +version = "0.116.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "anyio" }, + { name = "distro" }, + { name = "docstring-parser" }, + { name = "httpx" }, + { name = "jiter" }, + { name = "pydantic" }, + { name = "sniffio" }, + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/66/a2/d31f14e28d49bae983a3634e38dfb4b31c50110b5e403596c5c6a20b23f8/anthropic-0.116.0.tar.gz", hash = "sha256:5fc248fbb9fe03ef686f8a774f81586bca31a043260aab88b387ea3660f4a396", size = 949149, upload-time = "2026-07-02T19:08:10.534Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/c7/dd/2a1e81cf1b163acc340afc4ec74ed1d86f5eed1a809fabdeed3e0997b346/anthropic-0.116.0-py3-none-any.whl", hash = "sha256:6c0a7698e8d652455da3499978279bb2588c7264d0a35be3666009a4258c8256", size = 956896, upload-time = "2026-07-02T19:08:08.756Z" }, +] + [[package]] name = "anyio" version = "4.13.0" @@ -62,7 +81,7 @@ wheels = [ [[package]] name = "batchor" -version = "0.0.4" +version = "0.0.5" source = { editable = "." } dependencies = [ { name = "openai" }, @@ -76,6 +95,9 @@ dependencies = [ ] [package.optional-dependencies] +anthropic = [ + { name = "anthropic" }, +] gemini = [ { name = "google-cloud-storage" }, { name = "google-genai" }, @@ -97,6 +119,7 @@ docs = [ [package.metadata] requires-dist = [ + { name = "anthropic", marker = "extra == 'anthropic'", specifier = ">=0.71.0,<1" }, { name = "google-cloud-storage", marker = "extra == 'gemini'", specifier = ">=3.4.0,<4" }, { name = "google-genai", marker = "extra == 'gemini'", specifier = ">=1.55.0,<2" }, { name = "openai", specifier = ">=2.7.1" }, @@ -108,7 +131,7 @@ requires-dist = [ { name = "tiktoken", specifier = ">=0.12.0" }, { name = "typer", specifier = ">=0.16.0,<1" }, ] -provides-extras = ["gemini"] +provides-extras = ["anthropic", "gemini"] [package.metadata.requires-dev] dev = [ @@ -430,6 +453,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/12/b3/231ffd4ab1fc9d679809f356cebee130ac7daa00d6d6f3206dd4fd137e9e/distro-1.9.0-py3-none-any.whl", hash = "sha256:7bffd925d65168f85027d8da9af6bddab658135b840670a223589bc0c8ef02b2", size = 20277, upload-time = "2023-12-24T09:54:30.421Z" }, ] +[[package]] +name = "docstring-parser" +version = "0.18.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/e0/4d/f332313098c1de1b2d2ff91cf2674415cc7cddab2ca1b01ae29774bd5fdf/docstring_parser-0.18.0.tar.gz", hash = "sha256:292510982205c12b1248696f44959db3cdd1740237a968ea1e2e7a900eeb2015", size = 29341, upload-time = "2026-04-14T04:09:19.867Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/a7/5f/ed01f9a3cdffbd5a008556fc7b2a08ddb1cc6ace7effa7340604b1d16699/docstring_parser-0.18.0-py3-none-any.whl", hash = "sha256:b3fcbed555c47d8479be0796ef7e19c2670d428d72e96da63f3a40122860374b", size = 22484, upload-time = "2026-04-14T04:09:18.638Z" }, +] + [[package]] name = "execnet" version = "2.1.2"