Anthropic Message Batches API

These APIs call Anthropic's native Message Batches endpoint. They do not use the OpenAI-compatible Provider.

executionkit.batches.AnthropicBatchClient

AnthropicBatchClient(api_key: str, *, base_url: str = DEFAULT_ANTHROPIC_BASE_URL, timeout: float = DEFAULT_HTTP_TIMEOUT_SECONDS)

Minimal stdlib client for the Anthropic Message Batches API.

Exposes exactly the three calls the fan-out helpers need: create a batch, poll it, and fetch its results file. All transport goes through :meth:_http_raw — the single seam tests replace to run without a network.

Source code in executionkit/batches.py

def __init__(
    self,
    api_key: str,
    *,
    base_url: str = DEFAULT_ANTHROPIC_BASE_URL,
    timeout: float = DEFAULT_HTTP_TIMEOUT_SECONDS,
) -> None:
    if not api_key:
        raise ValueError("api_key must be a non-empty string")
    if not base_url.startswith(("https://", "http://")):
        raise ValueError(f"base_url must be http(s), got {base_url!r}")
    self.api_key = api_key
    self.base_url = base_url.rstrip("/")
    self.timeout = timeout

create_batch async

create_batch(requests: Sequence[Mapping[str, Any]]) -> dict[str, Any]

POST /v1/messages/batches with the given request entries.

Source code in executionkit/batches.py

async def create_batch(
    self, requests: Sequence[Mapping[str, Any]]
) -> dict[str, Any]:
    """``POST /v1/messages/batches`` with the given request entries."""
    if not requests:
        raise ValueError("cannot create a batch with no requests")
    payload = json.dumps({"requests": list(requests)}).encode("utf-8")
    raw = await asyncio.to_thread(
        self._http_raw, "POST", f"{self.base_url}/v1/messages/batches", payload
    )
    return self._decode_json_object(raw, context="create_batch")

fetch_results async

fetch_results(results_url: str) -> list[dict[str, Any]]

Fetch and parse the JSONL results file of an ended batch.

Source code in executionkit/batches.py

async def fetch_results(self, results_url: str) -> list[dict[str, Any]]:
    """Fetch and parse the JSONL results file of an ended batch."""
    if not results_url.startswith(("https://", "http://")):
        raise ProviderError(
            f"batch results_url has unexpected scheme: {results_url!r}"
        )
    raw = await asyncio.to_thread(self._http_raw, "GET", results_url, None)
    entries: list[dict[str, Any]] = []
    for line_number, line in enumerate(raw.decode("utf-8").splitlines(), start=1):
        if not line.strip():
            continue
        try:
            parsed = json.loads(line)
        except json.JSONDecodeError as exc:
            raise ProviderError(
                f"batch results line {line_number} is not valid JSON: {exc}"
            ) from exc
        if not isinstance(parsed, dict):
            raise ProviderError(
                f"batch results line {line_number} is not a JSON object"
            )
        entries.append(parsed)
    return entries

get_batch async

get_batch(batch_id: str) -> dict[str, Any]

GET /v1/messages/batches/{id} — the polling call.

Source code in executionkit/batches.py

async def get_batch(self, batch_id: str) -> dict[str, Any]:
    """``GET /v1/messages/batches/{id}`` — the polling call."""
    url = f"{self.base_url}/v1/messages/batches/{batch_id}"
    raw = await asyncio.to_thread(self._http_raw, "GET", url, None)
    return self._decode_json_object(raw, context="get_batch")

executionkit.batches.consensus_batch async

consensus_batch(client: AnthropicBatchClient, model: str, prompt: str, *, num_samples: int = 5, strategy: VotingStrategy | str = 'majority', temperature: float = DEFAULT_BATCH_TEMPERATURE, max_tokens: int = DEFAULT_MAX_TOKENS, poll_interval: float = DEFAULT_POLL_INTERVAL_SECONDS, timeout: float = DEFAULT_BATCH_TIMEOUT_SECONDS) -> PatternResult[str]

Consensus over num_samples identical prompts via one message batch.

The batch-transport twin of :func:executionkit.consensus: same prompt replication, same voting (:func:executionkit.engine.voting.tally_votes), same result shape — but the samples travel as a single Message Batches submission instead of num_samples concurrent live calls.

Anthropic-Batches-specific: unlike the provider-agnostic headline patterns (:func:executionkit.consensus, :func:executionkit.map_reduce, and friends — all of which accept any LLMProvider-conforming object), this function only talks to Anthropic's Message Batches API and requires an :class:AnthropicBatchClient and Anthropic API key — it is not usable with an arbitrary OpenAI-compatible endpoint.

Returns:

Name	Type	Description
A	PatternResult[str]	class:PatternResult whose value is the winning response and
	PatternResult[str]	score the agreement ratio.

Raises:

Type	Description
ConsensusFailedError	When strategy="unanimous" and responses are not all identical.
ProviderError	On transport failure, poll timeout, or any non-succeeded batch entry.
ValueError	If num_samples is less than 1.

Metadata

agreement_ratio (float): Fraction of samples matching the winner. unique_responses (int): Number of distinct response strings observed. tie_count (int): Number of responses tied for the top vote count. batch_id (str): The provider-assigned Message Batches id.

Source code in executionkit/batches.py

async def consensus_batch(
    client: AnthropicBatchClient,
    model: str,
    prompt: str,
    *,
    num_samples: int = 5,
    strategy: VotingStrategy | str = "majority",
    temperature: float = DEFAULT_BATCH_TEMPERATURE,
    max_tokens: int = DEFAULT_MAX_TOKENS,
    poll_interval: float = DEFAULT_POLL_INTERVAL_SECONDS,
    timeout: float = DEFAULT_BATCH_TIMEOUT_SECONDS,
) -> PatternResult[str]:
    """Consensus over ``num_samples`` identical prompts via one message batch.

    The batch-transport twin of :func:`executionkit.consensus`: same prompt
    replication, same voting (:func:`executionkit.engine.voting.tally_votes`),
    same result shape — but the samples travel as a single Message Batches
    submission instead of ``num_samples`` concurrent live calls.

    Anthropic-Batches-specific: unlike the provider-agnostic headline patterns
    (:func:`executionkit.consensus`, :func:`executionkit.map_reduce`, and
    friends — all of which accept any ``LLMProvider``-conforming object), this
    function only talks to Anthropic's Message Batches API and requires an
    :class:`AnthropicBatchClient` and Anthropic API key — it is not usable
    with an arbitrary OpenAI-compatible endpoint.

    Returns:
        A :class:`PatternResult` whose ``value`` is the winning response and
        ``score`` the agreement ratio.

    Raises:
        ConsensusFailedError: When ``strategy="unanimous"`` and responses
            are not all identical.
        ProviderError: On transport failure, poll timeout, or any
            non-succeeded batch entry.
        ValueError: If ``num_samples`` is less than 1.

    Metadata:
        agreement_ratio (float): Fraction of samples matching the winner.
        unique_responses (int): Number of distinct response strings observed.
        tie_count (int): Number of responses tied for the top vote count.
        batch_id (str): The provider-assigned Message Batches id.
    """
    if num_samples < 1:
        raise ValueError(f"num_samples must be >= 1, got {num_samples}")
    if isinstance(strategy, str):
        strategy = VotingStrategy(strategy)

    requests = [
        {
            "custom_id": f"consensus-{index}",
            "params": {
                "model": model,
                "max_tokens": max_tokens,
                "temperature": temperature,
                "messages": [user_message(prompt)],
            },
        }
        for index in range(num_samples)
    ]

    batch_id, entries, usage = await _run_batch(
        client, requests, poll_interval=poll_interval, timeout=timeout
    )
    contents = [
        _entry_or_missing(entries, f"consensus-{index}", batch_id).text
        for index in range(num_samples)
    ]
    tally = tally_votes(contents, strategy)

    return PatternResult[str](
        value=tally.winner,
        score=tally.agreement_ratio,
        cost=usage,
        metadata=MappingProxyType(
            {
                "agreement_ratio": tally.agreement_ratio,
                "unique_responses": tally.unique_responses,
                "tie_count": tally.tie_count,
                "batch_id": batch_id,
            }
        ),
    )

executionkit.batches.map_batch async

map_batch(client: AnthropicBatchClient, model: str, prompts: Sequence[str], *, max_tokens: int = DEFAULT_MAX_TOKENS, temperature: float | None = None, poll_interval: float = DEFAULT_POLL_INTERVAL_SECONDS, timeout: float = DEFAULT_BATCH_TIMEOUT_SECONDS) -> PatternResult[tuple[str, ...]]

Map every prompt to a completion via one message batch (fan-out only).

The "map" half of a map-reduce: one request per prompt, submitted as a single batch, with responses returned in prompt order regardless of the order the provider finished them. Callers reduce however they like — including with :func:executionkit.engine.voting.tally_votes for a consensus-style reduce over non-identical prompts.

Anthropic-Batches-specific: like :func:consensus_batch, this function only talks to Anthropic's Message Batches API — unlike the provider-agnostic headline patterns (:func:executionkit.map_reduce and friends), it is not usable with other OpenAI-compatible endpoints.

Returns:

Name	Type	Description
A	PatternResult[tuple[str, ...]]	class:PatternResult whose value is a tuple of responses,
	PatternResult[tuple[str, ...]]	index-aligned with prompts. score is None (no intrinsic
	PatternResult[tuple[str, ...]]	quality measure for a map).

Raises:

Type	Description
ProviderError	On transport failure, poll timeout, or any non-succeeded batch entry (strict all-or-nothing).
ValueError	If prompts is empty.

Metadata

batch_id (str): The provider-assigned Message Batches id. num_requests (int): Number of prompts submitted.

Source code in executionkit/batches.py

async def map_batch(
    client: AnthropicBatchClient,
    model: str,
    prompts: Sequence[str],
    *,
    max_tokens: int = DEFAULT_MAX_TOKENS,
    temperature: float | None = None,
    poll_interval: float = DEFAULT_POLL_INTERVAL_SECONDS,
    timeout: float = DEFAULT_BATCH_TIMEOUT_SECONDS,
) -> PatternResult[tuple[str, ...]]:
    """Map every prompt to a completion via one message batch (fan-out only).

    The "map" half of a map-reduce: one request per prompt, submitted as a
    single batch, with responses returned in prompt order regardless of the
    order the provider finished them. Callers reduce however they like —
    including with :func:`executionkit.engine.voting.tally_votes` for a
    consensus-style reduce over non-identical prompts.

    Anthropic-Batches-specific: like :func:`consensus_batch`, this function
    only talks to Anthropic's Message Batches API — unlike the
    provider-agnostic headline patterns (:func:`executionkit.map_reduce` and
    friends), it is not usable with other OpenAI-compatible endpoints.

    Returns:
        A :class:`PatternResult` whose ``value`` is a tuple of responses,
        index-aligned with ``prompts``. ``score`` is ``None`` (no intrinsic
        quality measure for a map).

    Raises:
        ProviderError: On transport failure, poll timeout, or any
            non-succeeded batch entry (strict all-or-nothing).
        ValueError: If ``prompts`` is empty.

    Metadata:
        batch_id (str): The provider-assigned Message Batches id.
        num_requests (int): Number of prompts submitted.
    """
    if not prompts:
        raise ValueError("prompts must be non-empty")

    requests = []
    for index, prompt in enumerate(prompts):
        params: dict[str, Any] = {
            "model": model,
            "max_tokens": max_tokens,
            "messages": [user_message(prompt)],
        }
        if temperature is not None:
            params["temperature"] = temperature
        requests.append({"custom_id": f"map-{index}", "params": params})

    batch_id, entries, usage = await _run_batch(
        client, requests, poll_interval=poll_interval, timeout=timeout
    )
    values = tuple(
        _entry_or_missing(entries, f"map-{index}", batch_id).text
        for index in range(len(prompts))
    )

    return PatternResult[tuple[str, ...]](
        value=values,
        score=None,
        cost=usage,
        metadata=MappingProxyType({"batch_id": batch_id, "num_requests": len(prompts)}),
    )

Read Anthropic Message Batches for configuration, polling behavior, and failure handling.