AIRegulation/nexus-claude-api
7
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor models and logging for nexus-claude-api

Browse Source 
- Update default models in config to use `claude-opus-4.6`.
- Introduce logging configuration to write logs to a file.
- Add correlation ID to error responses for better traceability.
- Implement diagnostics for summarizing message requests.
- Normalize legacy system messages in the API.
- Enhance tests to cover new logging and error handling features.
- Update README and documentation to reflect changes in model defaults and logging behavior.
This commit is contained in:
Guangfei.Zhao
2026-06-26 22:36:09 +08:00
parent 2851fa01cf
commit 0e98ce57d4
21 changed files with 573 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
.gitignore vendored
View File

@@ -10,3 +10,11 @@ __pycache__/
.env.*
!.env.example
nexus-claude-api.local.json
# agents
.omo
.codegraph
.agents
# logs
logs

30
AGENTS.md Normal file
View File

@@ -0,0 +1,30 @@
# AGENTS.md
## Repo shape
- Single Python package repo. Runtime code lives in `src/nexus_claude_api/`, tests in `tests/`, design/product docs in `docs/`.
- There is no monorepo/workspace layer, no task runner, and no repo-configured lint/typecheck/build command. Do not invent those steps in plans or completion reports.
## Commands that are actually supported
- Install dev dependencies: `uv sync --extra dev`
- Start the local proxy and print the Claude Code helper: `uv run nexus-claude-api start --port 4141 --claude-code`
- Fast config check without starting the server: `uv run nexus-claude-api start --dry-run`
- Run all tests: `uv run pytest`
- Run focused tests: `uv run pytest tests/test_cli.py` or `uv run pytest tests/test_routes.py -k messages_stream`
## Real entrypoints and change boundaries
- CLI entrypoint is `src/nexus_claude_api/cli.py`; `src/nexus_claude_api/__main__.py` just forwards to it.
- FastAPI app wiring is in `src/nexus_claude_api/server.py`; it stores `settings` and `nexus_client` on `app.state`.
- The main HTTP behavior is in `src/nexus_claude_api/routes/messages.py`. That file owns request validation, Anthropic→Bedrock translation, SSE streaming, and Anthropic-shaped error responses.
- Translation boundaries live under `src/nexus_claude_api/translators/`. Upstream Nexus/Bedrock transport lives in `src/nexus_claude_api/nexus_client.py`.
## Runtime/config facts worth preserving
- Credential lookup order is fixed in `Settings.from_values()`: `--api-key` -> `nexus-claude-api.local.json` -> `NEXUS_API_KEY` -> `AWS_BEARER_TOKEN_BEDROCK`.
- `nexus-claude-api.local.json` is gitignored and may exist locally; never commit it or overwrite it casually while testing.
- `NexusClient` writes `settings.api_key` back into `AWS_BEARER_TOKEN_BEDROCK` before building the boto3 Bedrock client. Keep that side effect in mind for tests and env-sensitive refactors.
- Defaults from `src/nexus_claude_api/config.py`: host `127.0.0.1`, port `4141`, endpoint `https://genai-nexus.api.corpinter.net`, main model `claude-opus-4.6`, small model `claude-opus-4.6`. Sonnet and Haiku aliases intentionally resolve to Opus for users who only have Opus access.
- The Claude Code helper from `src/nexus_claude_api/shell.py` intentionally sets `ANTHROPIC_AUTH_TOKEN='dummy'`. That placeholder is only for Claude Code compatibility; it is not the Nexus key.
## Repo-specific verification gotchas
- Tests rely on pytest `pythonpath = ["src"]` from `pyproject.toml`; keep the src-layout imports intact.
- `tests/test_cli.py` uses per-test temp workspaces under `.test-tmp/`. Logging writes to `logs/nexus-claude-api.log`. Both paths are gitignored.
- If you change request/error/logging behavior, re-run `uv run pytest tests/test_routes.py`. It covers SSE responses, correlation IDs, and the requirement that logs do not leak API keys, `Authorization`, or raw payload blobs.

3
README.md
View File

@@ -16,9 +16,12 @@ Use the printed Claude Code command in the same shell.
`ANTHROPIC_AUTH_TOKEN='dummy'` in the printed command is only a local Claude Code compatibility placeholder. Claude Code expects an Anthropic auth token variable to exist, but this local proxy does not validate it by default. It is not your Nexus key.
The printed command ends with `claude --model <configured-model>` so the current Claude Code process uses this proxy's configured model instead of any model saved in your global Claude settings. The helper does not edit `C:\Users\A200477427\.claude\settings.json` or any other global Claude Code configuration.
## Endpoints
- `GET /`
- `HEAD /`
- `GET /health`
- `GET /v1/models`
- `POST /v1/messages`

8
docs/PRD.md
View File

@@ -42,16 +42,14 @@ The first version will not include:
The local API exposes:
- `claude-sonnet-4.6`
- `claude-opus-4.6`
- `claude-haiku-4.5`
Defaults:
- Main model: `claude-sonnet-4.6`
- Small model: `claude-haiku-4.5`
- Main model: `claude-opus-4.6`
- Small model: `claude-opus-4.6`
`claude-sonnet-4.6` is the default because the AI Nexus documentation recommends it as the cost-effective default for most use cases.
The proxy defaults to Opus because this deployment is intended for users whose Nexus access is limited to `claude-opus-4.6`. Sonnet and Haiku aliases are accepted for compatibility and resolved to Opus before calling Nexus.
## User Stories

14
docs/REQUIREMENTS_DESIGN.md
View File

@@ -60,8 +60,8 @@ Options:
- `--port`, `-p`: default `4141`
- `--endpoint-url`: default `https://genai-nexus.api.corpinter.net`
- `--api-key`: optional; fallback to ignored local config, `NEXUS_API_KEY`, then `AWS_BEARER_TOKEN_BEDROCK`
- `--model`: default `claude-sonnet-4.6`
- `--small-model`: default `claude-haiku-4.5`
- `--model`: default `claude-opus-4.6`
- `--small-model`: default `claude-opus-4.6`
- `--claude-code`: print Claude Code launch command
- `--verbose`, `-v`: debug logging without secrets
@@ -95,15 +95,15 @@ Inbound authentication headers are accepted for compatibility but not validated
Public local model IDs:
- `claude-sonnet-4.6`
- `claude-opus-4.6`
- `claude-haiku-4.5`
Backend IDs are resolved through a mapping table. The initial default mapping keeps the same IDs, except common short aliases are supported:
Backend IDs are resolved through a mapping table. This deployment exposes Opus and maps common Sonnet/Haiku aliases to Opus for users whose Nexus access is limited to `claude-opus-4.6`:
- `claude-sonnet-4` -> `claude-sonnet-4.6`
- `claude-sonnet-4.6` -> `claude-opus-4.6`
- `claude-haiku-4.5` -> `claude-opus-4.6`
- `claude-sonnet-4` -> `claude-opus-4.6`
- `claude-opus-4` -> `claude-opus-4.6`
- `claude-haiku-4` -> `claude-haiku-4.5`
- `claude-haiku-4` -> `claude-opus-4.6`
If Nexus requires different backend IDs, update the mapping without changing Claude Code-facing model IDs.

11
src/nexus_claude_api/cli.py
View File

@@ -13,6 +13,7 @@ from nexus_claude_api.config import (
DEFAULT_SMALL_MODEL,
Settings,
)
from nexus_claude_api.logging_config import configure_logging
from nexus_claude_api.server import create_app
from nexus_claude_api.shell import generate_claude_code_powershell
@@ -73,8 +74,16 @@ def run_start(args: argparse.Namespace) -> int:
if args.dry_run:
return 0
log_file = configure_logging(verbose=settings.verbose)
print(f"Writing detailed logs to {log_file}")
app = create_app(settings)
uvicorn.run(app, host=settings.host, port=settings.port, log_level="debug" if settings.verbose else "info")
uvicorn.run(
app,
host=settings.host,
port=settings.port,
log_level="debug" if settings.verbose else "info",
access_log=True,
)
return 0

12
src/nexus_claude_api/config.py
View File

@@ -9,19 +9,19 @@ from pathlib import Path
DEFAULT_ENDPOINT_URL = "https://genai-nexus.api.corpinter.net"
DEFAULT_HOST = "127.0.0.1"
DEFAULT_PORT = 4141
DEFAULT_MODEL = "claude-sonnet-4.6"
DEFAULT_SMALL_MODEL = "claude-haiku-4.5"
DEFAULT_OPUS_MODEL = "claude-opus-4.6"
DEFAULT_MODEL = DEFAULT_OPUS_MODEL
DEFAULT_SMALL_MODEL = DEFAULT_OPUS_MODEL
LOCAL_CONFIG_FILENAME = "nexus-claude-api.local.json"
MODEL_ID_MAP = {
"claude-sonnet-4.6": "claude-sonnet-4.6",
"claude-opus-4.6": "claude-opus-4.6",
"claude-haiku-4.5": "claude-haiku-4.5",
"claude-sonnet-4": "claude-sonnet-4.6",
"claude-sonnet-4.6": DEFAULT_OPUS_MODEL,
"claude-haiku-4.5": DEFAULT_OPUS_MODEL,
"claude-sonnet-4": DEFAULT_OPUS_MODEL,
"claude-opus-4": "claude-opus-4.6",
"claude-haiku-4": "claude-haiku-4.5",
"claude-haiku-4": DEFAULT_OPUS_MODEL,
}

66
src/nexus_claude_api/diagnostics.py Normal file
View File

@@ -0,0 +1,66 @@
from __future__ import annotations
import base64
from collections import Counter
from typing import Any
from nexus_claude_api.config import resolve_backend_model
from nexus_claude_api.models import (
AnthropicImageBlock,
AnthropicMessage,
AnthropicMessagesRequest,
)
def summarize_messages_request(
payload: AnthropicMessagesRequest,
*,
correlation_id: str,
) -> dict[str, Any]:
content_types: Counter[str] = Counter()
images: list[dict[str, Any]] = []
for message in payload.messages:
_summarize_message(message, content_types=content_types, images=images)
return {
"correlation_id": correlation_id,
"model": payload.model,
"backend_model": resolve_backend_model(payload.model),
"stream": bool(payload.stream),
"message_count": len(payload.messages),
"content_block_types": dict(sorted(content_types.items())),
"images": images,
"tools": {
"count": len(payload.tools or []),
"names": [tool.name for tool in payload.tools or []],
},
"tool_choice_type": payload.tool_choice.type if payload.tool_choice else None,
}
def _summarize_message(
message: AnthropicMessage,
*,
content_types: Counter[str],
images: list[dict[str, Any]],
) -> None:
if isinstance(message.content, str):
content_types["text"] += 1
return
for block in message.content:
content_types[block.type] += 1
if isinstance(block, AnthropicImageBlock):
images.append(_summarize_image(block))
def _summarize_image(block: AnthropicImageBlock) -> dict[str, Any]:
summary: dict[str, Any] = {
"media_type": block.source.media_type,
}
try:
summary["byte_size"] = len(base64.b64decode(block.source.data, validate=True))
except Exception:
summary["byte_size"] = None
return summary

17
src/nexus_claude_api/errors.py
View File

@@ -23,15 +23,23 @@ def anthropic_error_response(
*,
status_code: int = 400,
error_type: str = "invalid_request_error",
correlation_id: str | None = None,
) -> JSONResponse:
error = {
"type": error_type,
"message": (
f"{message} [correlation_id={correlation_id}]"
if correlation_id
else message
),
}
if correlation_id:
error["correlation_id"] = correlation_id
return JSONResponse(
status_code=status_code,
content={
"type": "error",
"error": {
"type": error_type,
"message": message,
},
"error": error,
},
)
@@ -43,4 +51,3 @@ def map_http_exception(exc: HTTPException) -> JSONResponse:
status_code=exc.status_code,
error_type="invalid_request_error",
)

26
src/nexus_claude_api/logging_config.py Normal file
View File

@@ -0,0 +1,26 @@
from __future__ import annotations
import logging
from pathlib import Path
LOG_DIR = Path("logs")
LOG_FILE = LOG_DIR / "nexus-claude-api.log"
LOG_FORMAT = "%(asctime)s %(levelname)s %(name)s: %(message)s"
def configure_logging(*, verbose: bool = False, log_file: Path = LOG_FILE) -> Path:
log_file.parent.mkdir(parents=True, exist_ok=True)
formatter = logging.Formatter(LOG_FORMAT)
file_handler = logging.FileHandler(log_file, encoding="utf-8")
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(formatter)
app_logger = logging.getLogger("nexus_claude_api")
app_logger.handlers.clear()
app_logger.setLevel(logging.DEBUG if verbose else logging.INFO)
app_logger.addHandler(file_handler)
app_logger.propagate = False
return log_file

11
src/nexus_claude_api/models.py
View File

@@ -118,20 +118,9 @@ class CountTokensResponse(BaseModel):
SUPPORTED_MODELS = [
{
"id": "claude-sonnet-4.6",
"display_name": "Claude Sonnet 4.6",
"owned_by": "anthropic",
},
{
"id": "claude-opus-4.6",
"display_name": "Claude Opus 4.6",
"owned_by": "anthropic",
},
{
"id": "claude-haiku-4.5",
"display_name": "Claude Haiku 4.5",
"owned_by": "anthropic",
},
]

55
src/nexus_claude_api/nexus_client.py
View File

@@ -1,5 +1,6 @@
from __future__ import annotations
import logging
import os
from typing import Any
@@ -10,6 +11,9 @@ from nexus_claude_api.config import Settings
from nexus_claude_api.errors import NexusClaudeError
logger = logging.getLogger(__name__)
class NexusClient:
def __init__(self, settings: Settings) -> None:
if settings.api_key:
@@ -21,11 +25,20 @@ class NexusClient:
region_name="nexus",
)
def converse(self, request: dict[str, Any]) -> dict[str, Any]:
def converse(
self,
request: dict[str, Any],
*,
correlation_id: str | None = None,
) -> dict[str, Any]:
try:
return self._client.converse(**request)
except ClientError as exc:
raise _map_client_error(exc) from exc
raise _map_client_error(
exc,
operation="converse",
correlation_id=correlation_id,
) from exc
except BotoCoreError as exc:
raise NexusClaudeError(
"Failed to call Nexus Converse API",
@@ -33,12 +46,21 @@ class NexusClient:
error_type="api_error",
) from exc
def converse_stream(self, request: dict[str, Any]) -> Any:
def converse_stream(
self,
request: dict[str, Any],
*,
correlation_id: str | None = None,
) -> Any:
try:
response = self._client.converse_stream(**request)
return response.get("stream", [])
except ClientError as exc:
raise _map_client_error(exc) from exc
raise _map_client_error(
exc,
operation="converse_stream",
correlation_id=correlation_id,
) from exc
except BotoCoreError as exc:
raise NexusClaudeError(
"Failed to call Nexus Converse Stream API",
@@ -47,14 +69,33 @@ class NexusClient:
) from exc
def _map_client_error(exc: ClientError) -> NexusClaudeError:
def _map_client_error(
exc: ClientError,
*,
operation: str,
correlation_id: str | None = None,
) -> NexusClaudeError:
error = exc.response.get("Error", {})
code = error.get("Code", "")
message = error.get("Message", "Nexus API request failed")
status_code = int(exc.response.get("ResponseMetadata", {}).get("HTTPStatusCode", 502))
metadata = exc.response.get("ResponseMetadata", {})
status_code = int(metadata.get("HTTPStatusCode", 502))
request_id = metadata.get("RequestId")
if code in {"AccessDeniedException", "UnrecognizedClientException"}:
status_code = 403
elif code in {"ThrottlingException", "TooManyRequestsException"}:
status_code = 429
log_fields = {
"correlation_id": correlation_id,
"operation": operation,
"nexus_error_code": code,
"nexus_error_message": message,
"http_status": status_code,
"nexus_request_id": request_id,
}
logger.warning(
"nexus_client_error %s",
log_fields,
extra=log_fields,
)
return NexusClaudeError(message, status_code=status_code, error_type="api_error")

7
src/nexus_claude_api/routes/health.py
View File

@@ -1,6 +1,7 @@
from __future__ import annotations
from fastapi import APIRouter
from fastapi.responses import Response
router = APIRouter()
@@ -11,7 +12,11 @@ def root() -> dict[str, str]:
return {"status": "ok", "service": "nexus-claude-api"}
@router.head("/")
def root_head() -> Response:
return Response(status_code=200)
@router.get("/health")
def health() -> dict[str, str]:
return {"status": "ok"}

62
src/nexus_claude_api/routes/messages.py
View File

@@ -1,11 +1,14 @@
from __future__ import annotations
import logging
import uuid
from typing import Annotated
from fastapi import APIRouter, Depends, Request
from fastapi.responses import JSONResponse, Response, StreamingResponse
from pydantic import ValidationError
from nexus_claude_api.diagnostics import summarize_messages_request
from nexus_claude_api.errors import NexusClaudeError, anthropic_error_response
from nexus_claude_api.models import (
AnthropicMessagesRequest,
@@ -23,23 +26,64 @@ from nexus_claude_api.translators.stream import (
router = APIRouter()
logger = logging.getLogger(__name__)
def get_nexus_client(request: Request) -> NexusClient:
return request.app.state.nexus_client
def normalize_legacy_system_messages(raw: object) -> object:
if not isinstance(raw, dict):
return raw
messages = raw.get("messages")
if not isinstance(messages, list):
return raw
system_parts: list[object] = []
normalized_messages: list[object] = []
changed = False
for message in messages:
if isinstance(message, dict) and message.get("role") == "system":
system_parts.append(message.get("content", ""))
changed = True
else:
normalized_messages.append(message)
if not changed:
return raw
normalized = dict(raw)
normalized["messages"] = normalized_messages
if "system" not in normalized and system_parts:
normalized["system"] = system_parts[0] if len(system_parts) == 1 else system_parts
return normalized
@router.post("/v1/messages", response_model=None)
async def create_message(
request: Request,
client: Annotated[NexusClient, Depends(get_nexus_client)],
) -> Response:
correlation_id = uuid.uuid4().hex
try:
raw = await request.json()
raw = normalize_legacy_system_messages(await request.json())
payload = AnthropicMessagesRequest.model_validate(raw)
summary = summarize_messages_request(
payload,
correlation_id=correlation_id,
)
logger.info(
"anthropic_messages_request %s",
summary,
extra=summary,
)
bedrock_request = anthropic_to_bedrock_request(payload)
if payload.stream:
stream = client.converse_stream(bedrock_request)
stream = client.converse_stream(
bedrock_request,
correlation_id=correlation_id,
)
return StreamingResponse(
(
sse_frame(event)
@@ -51,26 +95,34 @@ async def create_message(
media_type="text/event-stream",
)
response = client.converse(bedrock_request)
response = client.converse(
bedrock_request,
correlation_id=correlation_id,
)
anthropic_response = bedrock_to_anthropic_response(
response,
model=payload.model,
)
return JSONResponse(content=anthropic_response.model_dump(exclude_none=True))
except ValidationError as exc:
return anthropic_error_response(str(exc), status_code=400)
return anthropic_error_response(
str(exc),
status_code=400,
correlation_id=correlation_id,
)
except NexusClaudeError as exc:
return anthropic_error_response(
exc.message,
status_code=exc.status_code,
error_type=exc.error_type,
correlation_id=correlation_id,
)
@router.post("/v1/messages/count_tokens")
async def count_tokens(request: Request) -> JSONResponse:
try:
raw = await request.json()
raw = normalize_legacy_system_messages(await request.json())
payload = CountTokensRequest.model_validate(raw)
response = CountTokensResponse(input_tokens=estimate_input_tokens(payload))
return JSONResponse(content=response.model_dump())

7
src/nexus_claude_api/server.py
View File

@@ -1,7 +1,5 @@
from __future__ import annotations
import logging
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.requests import Request
@@ -44,9 +42,4 @@ def create_app(
error_type=exc.error_type,
)
logging.basicConfig(
level=logging.DEBUG if resolved_settings.verbose else logging.INFO,
format="%(asctime)s %(levelname)s %(name)s: %(message)s",
)
return app

4
src/nexus_claude_api/shell.py
View File

@@ -14,7 +14,7 @@ def generate_claude_code_powershell(settings: Settings) -> str:
"ANTHROPIC_DEFAULT_HAIKU_MODEL": settings.small_model,
"DISABLE_NON_ESSENTIAL_MODEL_CALLS": "1",
"CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC": "1",
"CLAUDE_CODE_ENABLE_GATEWAY_MODEL_DISCOVERY": "1",
}
assignments = [f"$env:{key}='{value}'" for key, value in env.items()]
return "\n".join([*assignments, "claude"])
return "\n".join([*assignments, f"claude --model {settings.model}"])

20
tests/test_cli.py
View File

@@ -19,8 +19,24 @@ def test_claude_code_command() -> None:
command = generate_claude_code_powershell(settings)
assert "ANTHROPIC_BASE_URL" in command
assert "claude-sonnet-4.6" in command
assert command.endswith("claude")
assert "ANTHROPIC_AUTH_TOKEN='dummy'" in command
assert "CLAUDE_CODE_ENABLE_GATEWAY_MODEL_DISCOVERY='1'" in command
assert "claude-opus-4.6" in command
assert command.endswith("claude --model claude-opus-4.6")
def test_claude_code_command_uses_custom_model() -> None:
settings = Settings.from_values(
host="127.0.0.1",
port=4141,
api_key="test",
model="claude-opus-4.6",
require_api_key=False,
)
command = generate_claude_code_powershell(settings)
assert command.endswith("claude --model claude-opus-4.6")
def test_missing_api_key_fails(monkeypatch) -> None:

95
tests/test_diagnostics.py Normal file
View File

@@ -0,0 +1,95 @@
from __future__ import annotations
import base64
from nexus_claude_api.diagnostics import summarize_messages_request
from nexus_claude_api.models import AnthropicMessagesRequest
def test_text_request_summary_omits_prompt_text() -> None:
payload = AnthropicMessagesRequest.model_validate(
{
"model": "claude-opus-4.6",
"messages": [{"role": "user", "content": "secret prompt"}],
"max_tokens": 32,
}
)
summary = summarize_messages_request(payload, correlation_id="corr-1")
assert summary["correlation_id"] == "corr-1"
assert summary["model"] == "claude-opus-4.6"
assert summary["backend_model"] == "claude-opus-4.6"
assert summary["stream"] is False
assert summary["message_count"] == 1
assert summary["content_block_types"] == {"text": 1}
assert "secret prompt" not in repr(summary)
def test_image_request_summary_omits_base64_and_records_size() -> None:
image_data = base64.b64encode(b"image-bytes").decode("ascii")
payload = AnthropicMessagesRequest.model_validate(
{
"model": "claude-opus-4.6",
"messages": [
{
"role": "user",
"content": [
{"type": "text", "text": "describe"},
{
"type": "image",
"source": {
"type": "base64",
"media_type": "image/png",
"data": image_data,
},
},
],
}
],
"max_tokens": 32,
}
)
summary = summarize_messages_request(payload, correlation_id="corr-2")
assert summary["content_block_types"] == {"image": 1, "text": 1}
assert summary["images"] == [{"media_type": "image/png", "byte_size": 11}]
assert image_data not in repr(summary)
assert "image-bytes" not in repr(summary)
def test_tool_request_summary_omits_tool_inputs() -> None:
payload = AnthropicMessagesRequest.model_validate(
{
"model": "claude-opus-4.6",
"messages": [
{
"role": "assistant",
"content": [
{
"type": "tool_use",
"id": "toolu_1",
"name": "lookup_secret",
"input": {"api_key": "secret-tool-input"},
}
],
}
],
"tools": [
{
"name": "lookup_secret",
"input_schema": {"type": "object"},
}
],
"tool_choice": {"type": "tool", "name": "lookup_secret"},
"max_tokens": 32,
}
)
summary = summarize_messages_request(payload, correlation_id="corr-3")
assert summary["tools"] == {"count": 1, "names": ["lookup_secret"]}
assert summary["tool_choice_type"] == "tool"
assert "secret-tool-input" not in repr(summary)
assert "api_key" not in repr(summary)

57
tests/test_logging_config.py Normal file
View File

@@ -0,0 +1,57 @@
from __future__ import annotations
import logging
from nexus_claude_api.logging_config import configure_logging
def test_configure_logging_writes_app_logs_to_file_without_console(
tmp_path,
capsys,
) -> None:
log_file = tmp_path / "nexus-claude-api.log"
logger = logging.getLogger("nexus_claude_api.routes.messages")
app_logger = logging.getLogger("nexus_claude_api")
original_handlers = app_logger.handlers[:]
original_level = app_logger.level
original_propagate = app_logger.propagate
try:
configure_logging(verbose=False, log_file=log_file)
logger.info("detailed app request")
logger.debug("debug details hidden without verbose")
output = capsys.readouterr()
log_text = log_file.read_text(encoding="utf-8")
assert output.err == ""
assert "detailed app request" in log_text
assert "debug details hidden without verbose" not in log_text
finally:
app_logger.handlers.clear()
app_logger.handlers.extend(original_handlers)
app_logger.setLevel(original_level)
app_logger.propagate = original_propagate
def test_configure_logging_verbose_writes_debug_app_logs_to_file(
tmp_path,
) -> None:
log_file = tmp_path / "nexus-claude-api.log"
logger = logging.getLogger("nexus_claude_api.routes.messages")
app_logger = logging.getLogger("nexus_claude_api")
original_handlers = app_logger.handlers[:]
original_level = app_logger.level
original_propagate = app_logger.propagate
try:
configure_logging(verbose=True, log_file=log_file)
logger.debug("debug details stay in file")
assert "debug details stay in file" in log_file.read_text(encoding="utf-8")
finally:
app_logger.handlers.clear()
app_logger.handlers.extend(original_handlers)
app_logger.setLevel(original_level)
app_logger.propagate = original_propagate

126
tests/test_routes.py
View File

@@ -1,13 +1,18 @@
from __future__ import annotations
import logging
from botocore.exceptions import ClientError
from fastapi.testclient import TestClient
from nexus_claude_api.config import Settings
from nexus_claude_api.nexus_client import _map_client_error
from nexus_claude_api.server import create_app
class FakeNexusClient:
def converse(self, request: dict) -> dict:
def converse(self, request: dict, *, correlation_id: str | None = None) -> dict:
assert correlation_id
assert request["messages"][0]["content"][0]["text"] == "Hello"
return {
"ResponseMetadata": {"RequestId": "req-route"},
@@ -16,7 +21,8 @@ class FakeNexusClient:
"usage": {"inputTokens": 3, "outputTokens": 1},
}
def converse_stream(self, request: dict):
def converse_stream(self, request: dict, *, correlation_id: str | None = None):
assert correlation_id
assert request["messages"][0]["content"][0]["text"] == "Hello"
return [
{"contentBlockDelta": {"contentBlockIndex": 0, "delta": {"text": "Hi"}}},
@@ -25,9 +31,32 @@ class FakeNexusClient:
]
def client() -> TestClient:
class AccessDeniedNexusClient:
def converse(self, request: dict, *, correlation_id: str | None = None) -> dict:
raise _map_client_error(
ClientError(
{
"Error": {
"Code": "AccessDeniedException",
"Message": "not allowed",
},
"ResponseMetadata": {
"HTTPStatusCode": 403,
"RequestId": "req-denied",
},
},
"Converse",
),
operation="converse",
correlation_id=correlation_id,
)
def client(nexus_client: object | None = None) -> TestClient:
settings = Settings.from_values(api_key="test", require_api_key=False)
return TestClient(create_app(settings=settings, nexus_client=FakeNexusClient()))
return TestClient(
create_app(settings=settings, nexus_client=nexus_client or FakeNexusClient())
)
def test_health() -> None:
@@ -36,22 +65,24 @@ def test_health() -> None:
assert response.json() == {"status": "ok"}
def test_root_head() -> None:
response = client().head("/")
assert response.status_code == 200
assert response.content == b""
def test_models() -> None:
response = client().get("/v1/models")
assert response.status_code == 200
data = response.json()["data"]
assert {model["id"] for model in data} >= {
"claude-sonnet-4.6",
"claude-opus-4.6",
"claude-haiku-4.5",
}
assert [model["id"] for model in data] == ["claude-opus-4.6"]
def test_messages_non_stream() -> None:
response = client().post(
"/v1/messages",
json={
"model": "claude-sonnet-4.6",
"model": "claude-opus-4.6",
"messages": [{"role": "user", "content": "Hello"}],
"max_tokens": 32,
},
@@ -63,12 +94,82 @@ def test_messages_non_stream() -> None:
assert body["content"][0]["text"] == "Hi"
def test_messages_normalizes_legacy_system_role() -> None:
response = client().post(
"/v1/messages",
json={
"model": "claude-opus-4.6",
"messages": [
{"role": "system", "content": "You are helpful."},
{"role": "user", "content": "Hello"},
],
"max_tokens": 32,
},
)
assert response.status_code == 200
assert response.json()["content"][0]["text"] == "Hi"
def test_messages_upstream_error_returns_correlation_id() -> None:
response = client(AccessDeniedNexusClient()).post(
"/v1/messages",
json={
"model": "claude-opus-4.6",
"messages": [{"role": "user", "content": "Hello"}],
"max_tokens": 32,
},
)
assert response.status_code == 403
error = response.json()["error"]
assert error["type"] == "api_error"
assert error["correlation_id"]
assert f"correlation_id={error['correlation_id']}" in error["message"]
def test_client_error_mapping_logs_sanitized_details(caplog) -> None:
caplog.set_level(logging.WARNING, logger="nexus_claude_api.nexus_client")
mapped = _map_client_error(
ClientError(
{
"Error": {
"Code": "AccessDeniedException",
"Message": "model access denied",
},
"ResponseMetadata": {
"HTTPStatusCode": 400,
"RequestId": "req-log",
},
},
"Converse",
),
operation="converse",
correlation_id="corr-log",
)
assert mapped.status_code == 403
assert mapped.error_type == "api_error"
record = caplog.records[-1]
assert record.levelno == logging.WARNING
assert record.correlation_id == "corr-log"
assert record.operation == "converse"
assert record.nexus_error_code == "AccessDeniedException"
assert record.http_status == 403
assert record.nexus_request_id == "req-log"
log_text = caplog.text
assert "api_key" not in log_text
assert "Authorization" not in log_text
assert "aW1hZ2UtYnl0ZXM=" not in log_text
def test_messages_stream() -> None:
with client().stream(
"POST",
"/v1/messages",
json={
"model": "claude-sonnet-4.6",
"model": "claude-opus-4.6",
"messages": [{"role": "user", "content": "Hello"}],
"max_tokens": 32,
"stream": True,
@@ -86,11 +187,10 @@ def test_count_tokens() -> None:
response = client().post(
"/v1/messages/count_tokens",
json={
"model": "claude-sonnet-4.6",
"model": "claude-opus-4.6",
"messages": [{"role": "user", "content": "Hello"}],
},
)
assert response.status_code == 200
assert response.json()["input_tokens"] > 0

12
tests/test_translators.py
View File

@@ -11,7 +11,7 @@ from nexus_claude_api.translators.stream import bedrock_stream_to_anthropic_even
def test_text_request_translation() -> None:
payload = AnthropicMessagesRequest.model_validate(
{
"model": "claude-sonnet-4.6",
"model": "claude-opus-4.6",
"messages": [{"role": "user", "content": "Hello"}],
"system": "You are helpful.",
"max_tokens": 100,
@@ -21,7 +21,7 @@ def test_text_request_translation() -> None:
request = anthropic_to_bedrock_request(payload)
assert request["modelId"] == "claude-sonnet-4.6"
assert request["modelId"] == "claude-opus-4.6"
assert request["messages"] == [{"role": "user", "content": [{"text": "Hello"}]}]
assert request["system"] == [{"text": "You are helpful."}]
assert request["inferenceConfig"]["maxTokens"] == 100
@@ -32,7 +32,7 @@ def test_image_request_translation() -> None:
image_data = base64.b64encode(b"image-bytes").decode("ascii")
payload = AnthropicMessagesRequest.model_validate(
{
"model": "claude-sonnet-4.6",
"model": "claude-opus-4.6",
"messages": [
{
"role": "user",
@@ -63,7 +63,7 @@ def test_image_request_translation() -> None:
def test_tool_translation() -> None:
payload = AnthropicMessagesRequest.model_validate(
{
"model": "claude-sonnet-4.6",
"model": "claude-opus-4.6",
"messages": [
{"role": "user", "content": "Use a tool"},
{
@@ -119,7 +119,7 @@ def test_bedrock_response_translation() -> None:
"usage": {"inputTokens": 10, "outputTokens": 3},
}
translated = bedrock_to_anthropic_response(response, model="claude-sonnet-4.6")
translated = bedrock_to_anthropic_response(response, model="claude-opus-4.6")
assert translated.id == "req-1"
assert translated.content[0].type == "text"
@@ -137,7 +137,7 @@ def test_stream_translation() -> None:
]
translated = list(
bedrock_stream_to_anthropic_events(events, model="claude-sonnet-4.6")
bedrock_stream_to_anthropic_events(events, model="claude-opus-4.6")
)
assert translated[0]["type"] == "message_start"