ractogateway.anthropic_developer_kit.kit

Anthropic Claude Developer Kit — production-grade Claude interface.

Usage:

from ractogateway import anthropic_developer_kit as anth

kit = anth.AnthropicDeveloperKit(model="claude-sonnet-4-5-20250929", default_prompt=my_prompt)
response = kit.chat(anth.ChatConfig(user_message="Hello"))

for chunk in kit.stream(anth.ChatConfig(user_message="Hello")):
    print(chunk.delta.text, end="", flush=True)

Note: Anthropic does NOT have a native embedding API. Use OpenAI or Google kits for embeddings.

class ractogateway.anthropic_developer_kit.kit.AnthropicDeveloperKit(model='claude-sonnet-4-5-20250929', *, api_key=None, default_prompt=None, exact_cache=None, semantic_cache=None, router=None, truncator=None, tracer=None, metrics=None)

Bases: object

Complete Anthropic Claude developer kit — chat, streaming, and optional performance/cost optimisation middleware.

Parameters:

• model (str) – Claude model (e.g. "claude-sonnet-4-5-20250929", "claude-opus-4-6"). Use "auto" when a CostAwareRouter is provided — the router will select the model per-request.

• api_key (str | None) – Anthropic API key. Falls back to ANTHROPIC_API_KEY env var.

• default_prompt (RactoPrompt | None) – RACTO prompt used when ChatConfig.prompt is None.

• exact_cache (ExactMatchCache | None) – Optional ExactMatchCache.

• semantic_cache (SemanticCache | None) – Optional SemanticCache.

• router (CostAwareRouter | None) – Optional CostAwareRouter. Required when model="auto".

• truncator (TokenTruncator | None) – Optional TokenTruncator.

• tracer (RactoTracer | None) – Optional RactoTracer. Emits OpenTelemetry spans for every chat and stream call. Requires pip install ractogateway[telemetry].

• metrics (GatewayMetricsMiddleware | None) – Optional GatewayMetricsMiddleware. Records Prometheus metrics (latency, tokens, cost, cache hit/miss). Requires pip install ractogateway[prometheus].

provider: str = 'anthropic'

chat(config)

Synchronous chat completion with optional middleware pipeline.

Middleware order: truncate → exact cache → semantic cache → route model → API call → write caches → record telemetry.

Return type:

LLMResponse

async achat(config)

Async chat completion with optional middleware pipeline.

Return type:

LLMResponse

stream(config)

Synchronous streaming via Anthropic’s messages.stream().

Example:

for chunk in kit.stream(config):
    print(chunk.delta.text, end="", flush=True)
    if chunk.is_final:
        print(f"\nTokens: {chunk.usage}")

Return type:

Iterator[StreamChunk]

async astream(config)

Async streaming via Anthropic’s async messages.stream().

Return type:

AsyncIterator[StreamChunk]