feat(api): return 429 when the Claude usage limit is reached

README.md

@@ -85,6 +85,33 @@ curl -X POST http://localhost:8000/v1/chat/completions \
   }'
 ```
 
+## Error Responses
+
+Errors use the OpenAI error shape: `{"error": {"message", "type", "code"}}`.
+
+| Status | `error.code` | Meaning |
+| --- | --- | --- |
+| 400 | `model_not_supported` | Claude rejected every attempted model. |
+| 409 | `session_busy` | The session already has an active Claude process. |
+| 429 | `usage_limit_reached` | The Claude subscription usage limit is reached. |
+| 429 | `rate_limit_exceeded` | This gateway's own per-client rate limit was hit. |
+| 503 | `claude_unavailable` | Claude Code failed to start for another reason. |
+
+`usage_limit_reached` is reported separately from `claude_unavailable` so callers can
+fall back to another provider instead of retrying. When the CLI reports a reset time,
+the response adds a `Retry-After` header and a `reset_at` field:
+
+```json
+{
+  "error": {
+    "message": "Claude subscription usage limit reached.",
+    "type": "rate_limit_error",
+    "code": "usage_limit_reached",
+    "reset_at": "2027-03-20T09:46:40+00:00"
+  }
+}
+```
+
 ## Configuration
 
 Common settings are in `claude_code_api/core/config.py`:

claude_code_api/api/chat.py

@@ -2,6 +2,7 @@
 
 import hashlib
 import json
+from datetime import datetime, timezone
 from typing import Any, Dict, Optional, Tuple
 
 import structlog
@@ -12,7 +13,9 @@
 from claude_code_api.core.claude_manager import (
     ClaudeModelNotSupportedError,
     ClaudeSessionConflictError,
+    ClaudeUsageLimitError,
     create_project_directory,
+    raise_for_usage_limit,
 )
 from claude_code_api.core.session_manager import SessionManager
 from claude_code_api.models.claude import get_default_model, validate_claude_model
@@ -33,6 +36,8 @@
 )
 
 logger = structlog.get_logger()
+
+MAX_RETRY_AFTER_SECONDS = 86400
 router = APIRouter()
 
 CHAT_COMPLETION_RESPONSES = {
@@ -49,17 +54,25 @@
     },
     400: {"model": ErrorResponse},
     422: {"model": ErrorResponse},
+    429: {"model": ErrorResponse},
     503: {"model": ErrorResponse},
     500: {"model": ErrorResponse},
 }
 
 
 def _http_error(
-    status_code: int, message: str, error_type: str, code: str
+    status_code: int,
+    message: str,
+    error_type: str,
+    code: str,
+    extra: Optional[Dict[str, Any]] = None,
+    headers: Optional[Dict[str, str]] = None,
 ) -> HTTPException:
+    error: Dict[str, Any] = {"message": message, "type": error_type, "code": code}
+    if extra:
+        error.update(extra)
     return HTTPException(
-        status_code=status_code,
-        detail={"error": {"message": message, "type": error_type, "code": code}},
+        status_code=status_code, detail={"error": error}, headers=headers
     )
 
 
@@ -116,6 +129,33 @@ def _extract_prompts(request: ChatCompletionRequest) -> Tuple[str, str]:
     return user_prompt, system_prompt
 
 
+def _usage_limit_error(
+    error: ClaudeUsageLimitError, session_id: Optional[str] = None
+) -> HTTPException:
+    logger.warning(
+        "Claude usage limit reached",
+        session_id=session_id,
+        reset_at=error.reset_at.isoformat() if error.reset_at else None,
+    )
+    extra: Dict[str, Any] = {}
+    headers: Dict[str, str] = {}
+    if error.reset_at:
+        extra["reset_at"] = error.reset_at.isoformat()
+        retry_after = (error.reset_at - datetime.now(timezone.utc)).total_seconds()
+        if retry_after > 0:
+            headers["Retry-After"] = str(
+                min(int(retry_after) + 1, MAX_RETRY_AFTER_SECONDS)
+            )
+    return _http_error(
+        status.HTTP_429_TOO_MANY_REQUESTS,
+        "Claude subscription usage limit reached.",
+        "rate_limit_error",
+        "usage_limit_reached",
+        extra=extra or None,
+        headers=headers or None,
+    )
+
+
 async def _resolve_session(
     session_manager: SessionManager,
     request: ChatCompletionRequest,
@@ -147,6 +187,7 @@ async def _collect_non_streaming_response(
     project_id: str,
 ) -> Dict[str, Any]:
     messages, parser = await _gather_claude_messages(claude_process)
+    raise_for_usage_limit(getattr(claude_process, "terminal_error", None))
     _log_message_summary(messages)
 
     usage_summary = OpenAIConverter.calculate_usage(parser)
@@ -356,6 +397,8 @@ def _register_cli_session(cli_session_id: str):
                 "invalid_request_error",
                 "model_not_supported",
             ) from e
+        except ClaudeUsageLimitError as e:
+            raise _usage_limit_error(e, session_id) from e
         except Exception as e:
             logger.error(
                 "Failed to create Claude session", session_id=session_id, error=str(e)
@@ -403,6 +446,8 @@ def _register_cli_session(cli_session_id: str):
     except HTTPException:
         # Re-raise HTTP exceptions
         raise
+    except ClaudeUsageLimitError as e:
+        raise _usage_limit_error(e, session_id) from e
     except Exception as e:
         logger.error(
             "Unexpected error in chat completion",

claude_code_api/core/claude_manager.py

@@ -1,10 +1,13 @@
 """Claude Code process management."""
 
 import asyncio
+import contextlib
 import json
 import os
+import re
 import subprocess
 from collections import deque
+from datetime import datetime, timezone
 from typing import Any, AsyncGenerator, Callable, Dict, List, Optional
 
 import structlog
@@ -16,6 +19,8 @@
 
 logger = structlog.get_logger()
 
+READER_DRAIN_SECONDS = 1.0
+
 
 class ClaudeProcess:
     """Manages a single Claude Code process."""
@@ -39,6 +44,7 @@ def __init__(
         self._on_cli_session_id = on_cli_session_id
         self._on_end = on_end
         self.last_error: Optional[str] = None
+        self._result_error: Optional[str] = None
         self._stderr_tail: deque[str] = deque(maxlen=20)
 
     async def start(
@@ -49,6 +55,7 @@ async def start(
     ) -> bool:
         """Start Claude Code process and wait for completion."""
         self.last_error = None
+        self._result_error = None
         try:
             # Prepare real command - using exact format from working Claudia example
             cmd = [settings.claude_binary_path]
@@ -153,6 +160,8 @@ async def _read_output(self):
                 if not data:
                     continue
 
+                self._record_output_error(data)
+
                 # Extract Claude's session ID from the first message
                 if not claude_session_id and data.get("session_id"):
                     claude_session_id = data["session_id"]
@@ -176,6 +185,15 @@ async def _read_output(self):
             if self._on_end:
                 self._on_end(self)
 
+    def _record_output_error(self, data: Dict[str, Any]) -> None:
+        """Capture errors Claude reports on stdout instead of stderr."""
+        if data.get("type") != "result" or not data.get("is_error"):
+            return
+
+        result_text = data.get("result")
+        if isinstance(result_text, str) and result_text.strip():
+            self._result_error = result_text.strip()
+
     async def _read_error(self):
         """Read stderr from process."""
         try:
@@ -192,6 +210,20 @@ async def _read_error(self):
         except Exception as e:
             logger.error("Error reading stderr", error=str(e))
 
+    @property
+    def terminal_error(self) -> Optional[str]:
+        """Error Claude reported on stdout, if the run ended in one."""
+        return self._result_error
+
+    async def _drain_readers(self) -> None:
+        """Let buffered stdout and stderr be consumed before reporting an error."""
+        for task in (self._output_task, self._error_task):
+            if task and not task.done():
+                with contextlib.suppress(asyncio.TimeoutError, asyncio.CancelledError):
+                    await asyncio.wait_for(
+                        asyncio.shield(task), timeout=READER_DRAIN_SECONDS
+                    )
+
     async def _verify_startup(self) -> bool:
         """Detect early process failures so API can return actionable errors."""
         if not self.process:
@@ -203,10 +235,14 @@ async def _verify_startup(self) -> bool:
         while loop.time() < deadline:
             return_code = self.process.returncode
             if return_code is None:
+                if self._result_error:
+                    break
                 await asyncio.sleep(0.05)
                 continue
 
-            if return_code == 0:
+            await self._drain_readers()
+
+            if return_code == 0 and not self._result_error:
                 return True
 
             error_text = self._compose_process_error(return_code)
@@ -219,11 +255,24 @@ async def _verify_startup(self) -> bool:
             )
             return False
 
+        if self._result_error:
+            self.last_error = self._result_error
+            logger.error(
+                "Claude reported a terminal error",
+                session_id=self.session_id,
+                error=self._result_error,
+            )
+            return False
+
         return True
 
     def _compose_process_error(self, return_code: int) -> str:
-        if self._stderr_tail:
-            return f"Claude exited with code {return_code}: {' | '.join(self._stderr_tail)}"
+        details: List[str] = []
+        if self._result_error:
+            details.append(self._result_error)
+        details.extend(self._stderr_tail)
+        if details:
+            return f"Claude exited with code {return_code}: {' | '.join(details)}"
         return f"Claude exited with code {return_code}"
 
     async def get_output(self) -> AsyncGenerator[Dict[str, Any], None]:
@@ -315,6 +364,22 @@ class ClaudeModelNotSupportedError(ClaudeManagerError):
     """Raised when Claude rejects a requested model."""
 
 
+class ClaudeUsageLimitError(ClaudeManagerError):
+    """Raised when the Claude subscription usage limit is reached."""
+
+    def __init__(self, message: str, reset_at: Optional[datetime] = None):
+        super().__init__(message)
+        self.reset_at = reset_at
+
+
+def raise_for_usage_limit(error_message: Optional[str]) -> None:
+    """Raise when Claude reported an exhausted subscription."""
+    if error_message and _is_usage_limit_error(error_message):
+        raise ClaudeUsageLimitError(
+            error_message, reset_at=_extract_usage_limit_reset(error_message)
+        )
+
+
 def _is_model_rejection_error(error_message: str) -> bool:
     lowered = (error_message or "").lower()
     patterns = (
@@ -328,6 +393,30 @@ def _is_model_rejection_error(error_message: str) -> bool:
     return any(pattern in lowered for pattern in patterns)
 
 
+def _is_usage_limit_error(error_message: str) -> bool:
+    return "usage limit reached" in (error_message or "").lower()
+
+
+def _epoch_to_datetime(value: str) -> Optional[datetime]:
+    try:
+        return datetime.fromtimestamp(int(value), tz=timezone.utc)
+    except (OverflowError, OSError, ValueError):
+        return None
+
+
+def _extract_usage_limit_reset(error_message: str) -> Optional[datetime]:
+    message = error_message or ""
+    marker = message.lower().find("usage limit reached")
+    if marker >= 0:
+        message = message[marker:]
+
+    for match in re.finditer(r"(?:resets?_at[=:]\s*|\|)(\d{10})(?!\d)", message):
+        reset = _epoch_to_datetime(match.group(1))
+        if reset:
+            return reset
+    return None
+
+
 def _resolve_opus_45_fallback(model_id: Optional[str]) -> Optional[str]:
     if not model_id or not model_id.startswith("claude-opus-4-6-"):
         return None
@@ -473,6 +562,7 @@ async def _start_with_fallback_models(
                 return process
 
             last_error = process.last_error or last_error
+            raise_for_usage_limit(last_error)
             if not _is_model_rejection_error(last_error):
                 raise ClaudeProcessStartError(last_error)
 

claude_code_api/main.py

@@ -125,8 +125,14 @@ def custom_openapi():
 async def http_exception_handler(request, exc):
     """Custom handler for HTTP exceptions to support OpenAI error format."""
     if isinstance(exc.detail, dict) and "error" in exc.detail:
-        return JSONResponse(status_code=exc.status_code, content=exc.detail)
-    return JSONResponse(status_code=exc.status_code, content={"detail": exc.detail})
+        return JSONResponse(
+            status_code=exc.status_code, content=exc.detail, headers=exc.headers
+        )
+    return JSONResponse(
+        status_code=exc.status_code,
+        content={"detail": exc.detail},
+        headers=exc.headers,
+    )
 
 
 @app.exception_handler(RequestValidationError)

claude_code_api/models/openai.py

@@ -277,6 +277,9 @@ class ErrorDetail(BaseModel):
     message: str = Field(..., description="Human-readable error message")
     type: str = Field(..., description="Error type")
     code: Optional[str] = Field(None, description="Error code")
+    reset_at: Optional[str] = Field(
+        None, description="ISO timestamp when the usage limit resets"
+    )
 
 
 class ErrorResponse(BaseModel):