Shared Library

Overview#

anar_shared is the shared Python package used by all 13 FastAPI backends. It provides a unified interface for LLM providers, JWT authentication, production middleware, service-to-service communication, OpenTelemetry setup, and Guard telemetry. Version: 0.3.0.

Installation#

From any product backend directory:

uv pip install -e ../../../shared/python

For development with test dependencies:

cd shared/python
uv pip install -e ".[dev]"

Module Reference#

Production Hardening#

The harden() function applies the entire middleware stack and error handlers to a FastAPI app in a single call:

from fastapi import FastAPI
from anar_shared import harden, health_response

app = FastAPI()

harden(
    app,
    product="anar-chat",
    debug=False,
    heavy_paths=["/api/v1/chat/completions"],
    max_body_bytes=10 * 1024 * 1024,
)

@app.get("/health")
async def health():
    return health_response("anar-chat", version="0.1.0")

Middleware Stack (applied in order)#

Error Handlers#

register_error_handlers (called by harden) registers two handlers:

• Unhandled exceptions return 500 with {"error": "Internal server error"}
• Validation errors return 422 with {"error": "Validation error", "details": [...]}

LLM Abstraction#

The LLMProvider abstract class defines a consistent interface across all providers:

from anar_shared import LLMConfig, get_llm, get_llm_for_model

config = LLMConfig(
    provider="groq",
    groq_api_key="gsk_...",
    default_model="llama-3.3-70b-versatile",
)

llm = get_llm(config)
response = await llm.chat([
    {"role": "system", "content": "You are a helpful assistant."},
    {"role": "user", "content": "What is the capital of Oman?"},
])

Provider Classes#

LLMProvider Interface#

Every provider implements three methods:

async def chat(messages, model=None, *, temperature=0.7, max_tokens=2048) -> str
async def chat_stream(messages, model=None, *, temperature=0.7, max_tokens=2048) -> AsyncIterator[str]
async def vision(prompt, image_b64, *, model=None, detail="high", max_tokens=4096) -> str

Automatic Provider Resolution#

get_llm_for_model selects the correct provider based on the model name and automatically routes through the Gateway when configured:

llm = get_llm_for_model("allam-2-7b", config)
# If config.gateway_url is set → GatewayLLM
# Otherwise → GroqLLM (allam-2-7b maps to groq)

Streaming#

Use chat_stream for token-by-token output:

async for token in llm.chat_stream([
    {"role": "user", "content": "Explain quantum computing"}
]):
    print(token, end="", flush=True)

Service Client#

AnarServiceClient provides typed methods for calling other Anar products:

from anar_shared import AnarServiceClient, ServiceConfig

client = AnarServiceClient(ServiceConfig(
    guard_url="http://guard-backend:8002",
    docs_url="http://docs-backend:8008",
    auth_token="your-service-token",
))

scan = await client.guard_scan("Check this content for PII")
print(scan.safe, scan.risk_level, scan.categories)

pii = await client.guard_pii_detect("My ID is 784-1234-1234567-1")

doc = await client.docs_process(file_bytes, "contract.pdf")
print(doc.text, doc.confidence, doc.language)

Available Service Methods#

Structured Logging#

setup_logging configures JSON-formatted log output suitable for log aggregation:

from anar_shared import setup_logging

setup_logging(debug=False)

Log output format:

{
  "timestamp": "2026-02-16T10:30:45.123456+00:00",
  "level": "INFO",
  "module": "production",
  "message": "GET /api/v1/health 200 12.45ms",
  "request_id": "550e8400-e29b-41d4-a716-446655440000"
}

LLM Call Timer#

The llm_call_timer utility measures LLM call latency for telemetry:

from anar_shared import llm_call_timer

with llm_call_timer() as timer:
    response = await llm.chat(messages)

telemetry.record(
    model="llama-3.3-70b-versatile",
    latency_ms=timer.elapsed_ms,
)