Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions .env.example
Original file line number Diff line number Diff line change
Expand Up @@ -182,6 +182,8 @@ JWT_SECRET=your-jwt-secret-key
# Provider API Keys (Optional - if you want to use different keys for different providers)
OPENAI_API_KEY=your-openai-api-key
ANTHROPIC_API_KEY=your-anthropic-api-key
# Optional spares, used in order after a key is refused (docs/anthropic.md).
# ANTHROPIC_API_KEY_1=
# Required only for Anthropic keys that can act on multiple workspaces.
ANTHROPIC_WORKSPACE_ID=
CEREBRAS_API_KEY=your-cerebras-api-key
Expand Down
38 changes: 38 additions & 0 deletions docs/anthropic.md
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,44 @@ Set `ANTHROPIC_API_KEY` in the server environment or Cloudflare Worker secrets.
The Worker forwards it to the Container. Restart the server, or deploy the
updated Worker and Container, after changing configuration.

## Spare keys and failover

Add spare keys as `ANTHROPIC_API_KEY_1`, `ANTHROPIC_API_KEY_2`, and so on:

```bash
npx wrangler secret put ANTHROPIC_API_KEY_1
```

Requests use the first key that is not resting, in that order, so the spare
carries traffic only after the primary is refused. When the Claude API refuses a
key, the gateway rests it and sends the same request once on the next key. This
applies to unified routes, native `/anthropic/v1/*` routes, and the
`auto:intelligence` chain. Refusals that rest a key:

| Response | Rest |
| --- | --- |
| 400 `invalid_request_error`, "Your credit balance is too low…" | 1 hour |
| 400 `invalid_request_error`, "You have reached your specified … API usage limits" | 1 hour |
| 429 with `error.details.error_code` `enforced_spend_limit_reached` | 1 hour |
| 402 `billing_error` | 1 hour |
| 401 `authentication_error` | 5 minutes |
| 429 rate limit | `retry-after`, or 1 minute, for that model |

These rests apply with `MODEL_COOLDOWN_ENABLED=true`, the production setting,
and are capped by `MODEL_COOLDOWN_MAX_SECONDS` (default 3600). A 403 can be
specific to one model, so it and other 400 errors return to the caller without
trying another key. A rested key returns to service when its rest ends. If every key
is resting, the request fails with 429 `model_cooldown`. Responses from unified
routes carry `X-MultiLLM-Credential-Attempts` with the number of keys tried.

Keys from different Anthropic organizations do not share thinking signatures:
a signed thinking block replays only in the organization that created it.
Prompt-cache affinity keeps a conversation on the key that served it, so a
conversation moves to the spare only when its key is refused. Its next request
may then fail with an invalid `signature` error; drop the earlier thinking
blocks, or start a new conversation. `ANTHROPIC_WORKSPACE_ID` is sent with every
key, so leave it empty when the keys belong to different organizations.

For a key that can act on multiple Anthropic workspaces, also set
`ANTHROPIC_WORKSPACE_ID` to the intended workspace. The server sends it as
`anthropic-workspace-id` for generation and model discovery. Caller headers
Expand Down
2 changes: 2 additions & 0 deletions docs/deployment-cloudflare.md
Original file line number Diff line number Diff line change
Expand Up @@ -255,6 +255,8 @@ Provider secrets are optional and should be set only when used:

```bash
npx wrangler secret put OPENAI_API_KEY
npx wrangler secret put ANTHROPIC_API_KEY
npx wrangler secret put ANTHROPIC_API_KEY_1
npx wrangler secret put OPENROUTER_API_KEY
npx wrangler secret put OPENCODE_GO_API_KEY
npx wrangler secret put CODEX_EASY_API_KEY
Expand Down
65 changes: 65 additions & 0 deletions providers/anthropic.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
"""Direct Claude API authentication and model metadata."""

import json
from collections.abc import Mapping
from typing import Any

Expand All @@ -9,6 +10,70 @@
"anthropic-beta": "Anthropic-Beta",
"x-client-request-id": "X-Client-Request-ID",
}
# An account that ran out of credit or reached a spend limit stays refused until its owner
# acts, so its key rests for the longest cooldown instead of being retried every minute.
ANTHROPIC_REFUSAL_REST_SECONDS = 3600
_REFUSAL_BODY_LIMIT = 65_536
_SPEND_LIMIT_MESSAGES = (
"You have reached your specified API usage limits",
"You have reached your specified workspace API usage limits",
)


def _header(headers: Any, name: str) -> str:
getter = getattr(headers, "get", None)
if getter is None:
return ""
value = getter(name)
if value is None:
value = next(
(item for key, item in headers.items() if str(key).lower() == name.lower()),
None,
)
return value if isinstance(value, str) else ""


def anthropic_credential_refusal(response: Any) -> str | None:
"""Name the account-wide refusal in a Claude API error response, or return None.

Anthropic reports an empty credit balance and a spend limit set by the account owner
as 400 invalid_request_error, the same shape as a malformed request, so only the
message tells them apart. A tier spend cap is a 429 with a distinct error code. Reading
the small JSON error body caches it on the response, so relays still forward it.
"""
status = getattr(response, "status_code", None)
if status == 402:
return "billing_error"
if status not in (400, 429):
return None
headers = getattr(response, "headers", None) or {}
if "json" not in _header(headers, "Content-Type").lower():
return None
length = _header(headers, "Content-Length")
if length and (not length.isdigit() or int(length) > _REFUSAL_BODY_LIMIT):
return None
try:
body = response.content
except Exception: # A body that cannot be read is not evidence of a refusal.
return None
try:
error = json.loads(body).get("error") if len(body) <= _REFUSAL_BODY_LIMIT else None
except (AttributeError, TypeError, ValueError, RecursionError):
return None
if not isinstance(error, Mapping):
return None
if status == 429:
details = error.get("details")
code = details.get("error_code") if isinstance(details, Mapping) else None
return "spend_cap" if code == "enforced_spend_limit_reached" else None
message = error.get("message")
if error.get("type") != "invalid_request_error" or not isinstance(message, str):
return None
if message.startswith("Your credit balance is too low"):
return "credit_exhausted"
if message.startswith(_SPEND_LIMIT_MESSAGES):
return "spend_limit"
return None


def with_anthropic_headers(
Expand Down
25 changes: 21 additions & 4 deletions routes/provider_credentials.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@

from error_handlers import APIError
from services.credential_context import observation_context, selection_context
from services.credential_pool import is_key_rejection
from services.model_cooldown import ModelCooldownCapacity, ModelCooldownExhausted
from providers.nanogpt import is_nanogpt_paygo_rejection
from routes.auto_routes import AutoRouteCandidateUnavailable
Expand Down Expand Up @@ -60,6 +61,19 @@ def _is_nanogpt_routing_refusal(response) -> bool:
)


# Providers whose unified requests move to the next configured key after a key refusal.
ROTATING_PROVIDERS = frozenset({"nanogpt", "anthropic"})


def is_anthropic_key_refusal(response) -> bool:
"""The Claude API refused the key (auth, billing, credit, spend limit or rate limit),
so the request generated nothing and may run on the next key. Dispatch already rested
the refused key, and the next selection skips it."""
return bool(getattr(response, "multillm_credential_refusal", None)) or is_key_rejection(
getattr(response, "status_code", None)
)


def _record_nanogpt_token_result(
app,
provider: str,
Expand Down Expand Up @@ -96,9 +110,9 @@ def _request_with_provider_token_rotation(
billing_refusal_is_routing: bool = False,
*, model=None, quota_bucket=None,
):
"""Send once per usable NanoGPT key after definite pre-generation failures."""
"""Send once per usable NanoGPT or Claude API key after definite pre-generation failures."""
attempt_limit = 1
if provider == "nanogpt":
if provider in ROTATING_PROVIDERS:
attempt_limit = max(1, len(auth_service_cls.get_api_keys(provider)))

attempted_tokens: set[str] = set()
Expand Down Expand Up @@ -147,7 +161,10 @@ def _request_with_provider_token_rotation(
# Every key would refuse this suffix identically, so rotating on
# would only surface an unrelated key's error to the caller.
break
if provider != "nanogpt" or not is_nanogpt_credential_rejection(
if provider == "anthropic":
if not is_anthropic_key_refusal(response):
break
elif provider != "nanogpt" or not is_nanogpt_credential_rejection(
response.status_code
):
break
Expand All @@ -162,6 +179,6 @@ def _add_credential_attempt_headers(
provider: str,
attempts: int,
) -> Response:
if provider == "nanogpt":
if provider in ROTATING_PROVIDERS:
response.headers["X-MultiLLM-Credential-Attempts"] = str(attempts)
return response
20 changes: 20 additions & 0 deletions routes/proxy.py
Original file line number Diff line number Diff line change
Expand Up @@ -51,8 +51,10 @@
stream_upstream_response,
)
from routes.dashboard_chat import register_dashboard_chat_routes
from routes.provider_credentials import is_anthropic_key_refusal
from services.nanogpt_key_pool import NanoGPTKeyPool, NanoGPTKeyPoolExhausted
from services.credential_context import dispatch_context, observation_context, proxy_selection_context
from services.model_cooldown import ModelCooldownCapacity, ModelCooldownExhausted
from services.model_cooldown import settings as cooldown_settings
from services.nanogpt_speed_breaker import NanoGPTSpeedBreaker
from services.provider_access_policy import provider_route_scope
Expand Down Expand Up @@ -445,6 +447,24 @@ def send_to_url(target_url: str):
else:
response = send_to_url(url)

if api_provider == "anthropic":
# A refused Claude API key generated nothing: send once on each other key.
# Dispatch rested the refused key, so the next selection skips it.
attempted_tokens = {auth_token}
while is_anthropic_key_refusal(response):
try:
next_token = auth_service_cls.get_api_key(api_provider, **credential_scope)
except (ModelCooldownExhausted, ModelCooldownCapacity):
break
if not next_token or next_token in attempted_tokens:
break
attempted_tokens.add(next_token)
response.close()
headers = proxy_service_cls.prepare_headers(
request.headers, api_provider, next_token, upstream_path=path,
)
response = send_to_url(url)

if speed_routed_body is not None and is_nanogpt_paygo_rejection(
getattr(response, "status_code", None),
getattr(response, "content", None),
Expand Down
2 changes: 2 additions & 0 deletions services/auth_primitives.py
Original file line number Diff line number Diff line change
Expand Up @@ -72,6 +72,8 @@ def provider_credential_env_names(provider: str) -> tuple[str, ...]:
)
if provider == "groq":
return ("GROQ_API_KEY", "GROQ_API_KEY_1", "GROQ_API_KEY_2")
if provider == "anthropic":
return ("ANTHROPIC_API_KEY", "ANTHROPIC_API_KEY_1")
if provider == "nanogpt":
return (
"NANOGPT_API_KEY",
Expand Down
24 changes: 12 additions & 12 deletions services/credential_pool.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
"""Several keys for one provider: use the first that is not resting, and rest refused ones.

Codex Everywhere pools accept a base key and numbered spares (`<BASE>`, `<BASE>_1`,
`<BASE>_2`, ...). Each request uses the first key in that order that is not resting. A key
Codex Everywhere and Claude API pools accept a base key and numbered spares (`<BASE>`,
`<BASE>_1`, `<BASE>_2`, ...). Each request uses the first key in that order that is not resting. A key
the provider refuses (401, 402 or 403, which includes an empty balance) rests for five
minutes and a rate-limited key (429) for one minute, so traffic moves to the next key and
returns once the rest ends. State is process-local.
Expand Down Expand Up @@ -30,6 +30,7 @@
POOLED_KEY_ENVS = {
**{pool.provider: pool.credential_env for pool in CODEX_EVERYWHERE_POOLS},
"ce-image": "CODEX_EVERYWHERE_API_KEY_GPT_IMAGE",
"anthropic": "ANTHROPIC_API_KEY",

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Preserve credential affinity on native and explicit Claude routes

When keys belong to different organizations, a conversation started on the spare while the primary is resting will silently switch back to the primary once its cooldown expires. The advertised affinity protection does not cover native /anthropic/v1/messages or explicit unified Anthropic requests: request_scope() in services/prompt_cache_affinity.py only creates scopes for unified auto: selectors, and prefer_credential() returns nothing without an active scope and candidate. Consequently, even with production affinity enabled, replaying the spare organization's signed thinking blocks after the primary recovers produces an invalid-signature 400, which is not retried. Extend conversation-to-credential binding to these routes so recovery does not move an otherwise healthy conversation between organizations.

Useful? React with 👍 / 👎.

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Keep configuration checks independent of Claude pool admission

With production's POOL_RESET_SCHEDULING_ENABLED=true, exhausting the configured Claude keys now also makes /admin/auto-routes fail with 429. Adding Anthropic to this pool makes AuthService.get_api_key("anthropic") raise ModelCooldownExhausted when every key is resting, but routes/auto_routes.py::_provider_is_configured() calls that method without catching cooldown errors. _admin_payload() checks every registered provider, so both GET and PUT responses fail even when the administrator is managing routes unrelated to Anthropic. Use configured-key enumeration rather than dispatch admission for this configuration check, or handle cooldown exhaustion while retaining the configured status.

Useful? React with 👍 / 👎.

}


Expand Down Expand Up @@ -264,21 +265,20 @@ def record_headers(
usage_windows: list[Mapping] | None = None,
now: float | None = None,
) -> None:
"""Record the result for the bearer key a request carried."""
"""Record the result for the bearer or X-Api-Key credential a request carried."""
if provider not in POOLED_KEY_ENVS:
return
authorization = next(
(
value
for name, value in headers.items()
if name.lower() == "authorization"
),
"",
)
sent = {name.lower(): value for name, value in headers.items()}
authorization = sent.get("authorization", "")
if isinstance(authorization, str) and authorization.startswith("Bearer "):
key = authorization[len("Bearer ") :].strip()
else:
api_key = sent.get("x-api-key")
key = api_key.strip() if isinstance(api_key, str) else ""
if key:
cls.record(
provider,
authorization[len("Bearer ") :].strip(),
key,
status,
model=model,
quota_bucket=quota_bucket,
Expand Down
4 changes: 3 additions & 1 deletion services/intelligence_gateway.py
Original file line number Diff line number Diff line change
Expand Up @@ -227,7 +227,9 @@ def events(self):
self.unresolved = False
last_error = rejection(head)
# A refused key generated nothing, so the same model may run on the next key.
self._key_refused = is_key_rejection(head.status_code)
self._key_refused = is_key_rejection(head.status_code) or getattr(
head, "credential_refused", False
)
if self.request.explicit and not (self._key_refused and spare_keys):
break
if not self.escalations:
Expand Down
6 changes: 5 additions & 1 deletion services/intelligence_transport.py
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,9 @@ def remaining(deadline, cancelled):
class ResponseHead:
status_code: int
headers: dict
# Dispatch classified the error as an account-wide key refusal (for example an empty
# Claude API credit balance reported as 400), so the next key may run the same model.
credential_refused: bool = False


class Exchange:
Expand Down Expand Up @@ -119,7 +122,8 @@ def _run(self):
for name in ("Content-Type", "Retry-After")
if name in response.headers
}
self._put(("head", ResponseHead(response.status_code, headers)))
refused = bool(getattr(response, "multillm_credential_refusal", None))
self._put(("head", ResponseHead(response.status_code, headers, refused)))
size = 0
# Definite rejections need no error-body read, which could otherwise
# disclose secrets or postpone a safe fallback.
Expand Down
1 change: 1 addition & 0 deletions services/managed_turn.py
Original file line number Diff line number Diff line change
Expand Up @@ -278,6 +278,7 @@ def __init__(self, response, upstream):
self.headers = CaseInsensitiveDict(response.headers)
self.response, self.upstream = response, upstream
self.raw = None
self.multillm_credential_refusal = getattr(upstream, "multillm_credential_refusal", None)

def iter_content(self, **kwargs):
yield from self.response.response
Expand Down
23 changes: 19 additions & 4 deletions services/proxy_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,11 @@
from services.transport_policy import RAW_PASSTHROUGH_PROVIDERS
from services.secret_firewall import protect_body
from providers.cline_pass import cline_completion_payload
from providers.anthropic import with_anthropic_headers
from providers.anthropic import (
ANTHROPIC_REFUSAL_REST_SECONDS,
anthropic_credential_refusal,
with_anthropic_headers,
)
from services.provider_request_headers import filter_upstream_headers
from providers.codex_everywhere import (
CODEX_EVERYWHERE_ANTHROPIC_PROVIDERS,
Expand Down Expand Up @@ -3325,9 +3329,20 @@ def make_request(
)
from services.credential_context import dispatch_context, observation_context
from services.pool_reset_schedule import response_usage_observation
CredentialPool.record_headers(api_provider, headers, response.status_code,
**(cooldown_context or dispatch_context(api_provider, data, url)),
**observation_context(response), **response_usage_observation(response))
observation = {**observation_context(response), **response_usage_observation(response)}
status = response.status_code
if api_provider == "anthropic" and (refusal := anthropic_credential_refusal(response)):
# The account, not this request, was refused: rest its key so the
# rotation loops and later requests move to the next configured key.
response.multillm_credential_refusal = refusal
status = 402
observation.update(
outcome=UpstreamOutcome(False, "neutral", "rejected", f"anthropic_{refusal}"),
credential_wide_auth=True,
retry_after_seconds=ANTHROPIC_REFUSAL_REST_SECONDS,
)
CredentialPool.record_headers(api_provider, headers, status,
**(cooldown_context or dispatch_context(api_provider, data, url)), **observation)
Comment on lines +3344 to +3345

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Honor pooled credentials at the policy-bound dispatch boundary

For accounts with an administrator-owned data policy, such as official_only=true, services/provider_dispatch.py::_credential() replaces the selected X-Api-Key with operator_credential("anthropic"), which reads only ANTHROPIC_API_KEY and ignores the pool. Therefore, after the primary is refused, selecting a spare still sends the exhausted primary again. This recording call then attributes that refusal to the originally selected spare, cooling down a healthy credential that was never sent and potentially exhausting the pool for other users too. Update the trusted credential-resolution path to honor eligible operator-configured pool keys, and record the credential actually dispatched.

Useful? React with 👍 / 👎.

return response

# Check if this is a streaming request
Expand Down
7 changes: 7 additions & 0 deletions services/upstream_transport.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,13 @@ def iter_stream_content(response: requests.Response, *, pii_context=None) -> Ite


def _stream_content(response: requests.Response) -> Iterator[bytes]:
if getattr(response, "_content_consumed", False) and isinstance(
getattr(response, "_content", None), bytes
):
# The body was already read (an error body classified at dispatch); the socket
# is drained, so replay the cached bytes.
yield from response.iter_content(chunk_size=64 * 1024)
return
raw_response = getattr(response, "raw", None)
raw_read1 = getattr(raw_response, "read1", None)
if callable(raw_read1):
Expand Down
Loading
Loading