majordomo-llm

A unified Python interface for multiple LLM providers with automatic cost tracking, retry logic, and structured output support.

Features

• Unified API - Same interface for OpenAI, Anthropic (Claude), Google Gemini, DeepSeek, Cohere, Amazon Bedrock, and the open-weight inference platforms (Fireworks, Together, Baseten, Nebius, DeepInfra, Moonshot, Novita)
• Streaming - Real-time token-by-token output via get_response_stream() with async iteration
• Cost Tracking - Automatic calculation of input/output token costs per request
• Structured Outputs - Native support for Pydantic models and raw JSON Schema dicts
• Automatic Retries - Built-in exponential backoff retry logic using tenacity
• Automatic Fallback - Cascade across providers with LLMCascade for resilience
• Optimal Routing - The majordomo provider lets the Majordomo gateway pick the best backend for a canonical open-weight model at request time, with cost resolved from the routed backend
• Request Logging - Optional async logging to PostgreSQL/MySQL/SQLite with S3 or local file storage for request/response bodies
• API Key Tracking - Log hashed API keys and optional aliases for usage attribution
• Async First - Fully async/await compatible for high-performance applications
• Type Safe - Complete type annotations and py.typed marker for IDE support

Installation

pip install majordomo-llm

Or with uv:

uv add majordomo-llm

Optional: Request Logging

To enable request logging to PostgreSQL, MySQL, or S3:

pip install majordomo-llm[logging]

Quick Start

Basic Text Response

import asyncio
from majordomo_llm import get_llm_instance

async def main():
    # Create an LLM instance
    llm = get_llm_instance("anthropic", "claude-sonnet-4-20250514")

    # Get a response
    response = await llm.get_response(
        user_prompt="What is the capital of France?",
        system_prompt="You are a helpful geography assistant.",
    )

    print(response.content)
    print(f"Tokens: {response.input_tokens} in, {response.output_tokens} out")
    print(f"Cost: ${response.total_cost:.6f}")

asyncio.run(main())

JSON Response

response = await llm.get_json_response(
    user_prompt="List the top 3 largest countries by area as JSON",
    system_prompt="Respond with valid JSON only.",
)

# response.content is a parsed Python dict
for country in response.content["countries"]:
    print(country["name"])

Streaming

stream = await llm.get_response_stream(
    user_prompt="Explain quantum computing",
    system_prompt="Be concise.",
)

async for chunk in stream:
    print(chunk, end="", flush=True)

print(f"\nCost: ${stream.usage.total_cost:.6f}")

# Or collect the full response:
stream = await llm.get_response_stream("Summarize this document...")
response = await stream.collect()  # Returns an LLMResponse
print(response.content)

Structured Output with Pydantic

from pydantic import BaseModel

class CountryInfo(BaseModel):
    name: str
    capital: str
    population: int
    area_km2: float

response = await llm.get_structured_json_response(
    response_model=CountryInfo,
    user_prompt="Give me information about Japan",
)

# response.content is a validated CountryInfo instance
country = response.content
print(f"{country.name}: {country.capital}, pop. {country.population:,}")

Structured Output with Raw JSON Schema

import json

schema = {
    "type": "object",
    "properties": {
        "name": {"type": "string"},
        "capital": {"type": "string"},
        "population": {"type": "integer"},
    },
    "required": ["name", "capital", "population"],
}

response = await llm.get_json_schema_response(
    user_prompt="Give me information about Japan",
    response_schema=schema,
    schema_name="CountryInfo",
)

# response.content is canonical JSON: sorted keys, no extra whitespace
country = json.loads(response.content)
print(country["capital"])

Configuration

Environment Variables

Set API keys for the providers you want to use:

# OpenAI
export OPENAI_API_KEY="sk-..."

# Anthropic (Claude)
export ANTHROPIC_API_KEY="sk-ant-..."

# Google Gemini
export GEMINI_API_KEY="..."

# DeepSeek
export DEEPSEEK_API_KEY="sk-..."

# Cohere
export CO_API_KEY="..."

# Open-weight inference platforms
export FIREWORKS_API_KEY="..."
export TOGETHER_API_KEY="..."
export BASETEN_API_KEY="..."
export NEBIUS_API_KEY="..."
export DEEPINFRA_API_KEY="..."
export MOONSHOT_API_KEY="..."
export NOVITA_API_KEY="..."

For local development, copy .env.example to .env and fill in your keys. Never commit .env.

Available Models

OpenAI

• gpt-5.5
• gpt-5.4, gpt-5.4-mini, gpt-5.4-nano, gpt-5.4-pro
• gpt-5, gpt-5-mini, gpt-5-nano
• gpt-4.1, gpt-4.1-mini, gpt-4.1-nano
• o3, o4-mini

Anthropic

• claude-opus-4-6, claude-sonnet-4-6
• claude-opus-4-5-20251101, claude-sonnet-4-5-20250929, claude-haiku-4-5-20251001
• claude-opus-4-1-20250805, claude-opus-4-20250514, claude-sonnet-4-20250514

Gemini

• gemini-3.1-pro-preview, gemini-3-flash-preview, gemini-3.1-flash-lite-preview
• gemini-2.5-pro, gemini-2.5-flash, gemini-2.5-flash-lite

DeepSeek

• deepseek-v4-flash, deepseek-v4-pro
• deepseek-chat, deepseek-reasoner

Cohere

• command-a-03-2025, command-r-plus-08-2024
• command-r-08-2024, command-r7b-12-2024

Open-Weight Inference Platforms

The same open-weight models are served by several platforms at different prices; pick a provider directly, or stack them behind an LLMCascade for failover.

Notes:

• Model IDs are case-sensitive and passed through verbatim, and the spelling varies by platform. Baseten writes Inkling lowercase (thinkingmachines/inkling) where Together and DeepInfra use thinkingmachines/Inkling; Novita uses deepseek/deepseek-v4-pro where the HF-style platforms use deepseek-ai/DeepSeek-V4-Pro.
• Capability flags are per host, not per model, and were measured against the live APIs rather than taken from vendor metadata. Some deployments pin their sampling parameters (Moonshot's Kimi SKUs require temperature=1/top_p=0.95; Baseten's Kimi-K3 requires top_p=0.95) and some reject or ignore strict json_schema. Those models carry supports_temperature_top_p: false or supports_structured_outputs: false in llm_config.yaml.
• Nebius publishes no cached-token rate, so cached reads bill at input_cost there (as with Fireworks and Together). Its rates come from GET /v1/models?verbose=true, which is also the quickest way to re-check them.
• Moonshot is the first-party Kimi vendor. The configured rates are the USD international platform; mainland-China accounts bill separately in RMB via https://api.moonshot.cn/v1 (reachable by passing it as base_url).

Majordomo (Optimal Routing)

Canonical open-weight models routed to the optimal backend by the Majordomo gateway (see Optimal Routing):

• deepseek-v4-pro, kimi-k2.6, kimi-k3
• glm-5.1, glm-5.2, inkling

Deprecated Model Handling

If you pass a deprecated model to get_llm_instance(), it is automatically replaced with the provider-recommended replacement and a warning is logged. The response object includes a deprecation_warning field so you can detect this in your application:

llm = get_llm_instance("openai", "gpt-4o")  # deprecated → auto-replaced with gpt-4.1

response = await llm.get_response("Hello!")
if response.deprecation_warning:
    print(response.deprecation_warning)
    # "Model 'gpt-4o' for provider 'openai' is deprecated.
    #  Automatically replaced with 'gpt-4.1'."

See the deprecated_models section in llm_config.yaml for the full mapping.

API Reference

Factory Functions

get_llm_instance(provider: str, model: str) -> LLM

Create an LLM instance for the specified provider and model.

from majordomo_llm import get_llm_instance

llm = get_llm_instance("openai", "gpt-4.1")

LLM Methods

All LLM instances support these async methods:

get_response(user_prompt, system_prompt=None, temperature=0.3, top_p=1.0) -> LLMResponse

Get a plain text response.

get_json_response(user_prompt, system_prompt=None, temperature=0.3, top_p=1.0) -> LLMJSONResponse

Get a JSON response (automatically parsed).

get_response_stream(user_prompt, system_prompt=None, temperature=0.3, top_p=1.0) -> LLMStreamResponse

Get a streaming text response. Yields chunks via async iteration; usage metrics are available after the stream completes.

get_structured_json_response(response_model, user_prompt, system_prompt=None, temperature=0.3, top_p=1.0) -> LLMStructuredResponse

Get a response validated against a Pydantic model.

get_json_schema_response(user_prompt, response_schema, system_prompt=None, schema_name="Response", schema_description=None, temperature=0.3, top_p=1.0) -> LLMResponse

Get a response validated against a raw JSON Schema dict. response.content is canonical JSON.

Sampling Parameters

temperature and top_p are optional and unset by default. They are sent only when you pass them, so each provider applies its own documented default:

await llm.get_response("Hello")                    # neither parameter is sent
await llm.get_response("Hello", temperature=0.2)   # only temperature is sent

Models whose deployment rejects these parameters never receive them — that covers every current OpenAI and Anthropic flagship, plus deployments that pin their values (Moonshot's Kimi SKUs require temperature=1 / top_p=0.95). Those models set supports_temperature_top_p: false in llm_config.yaml. Passing a value to one of them logs a warning and the call proceeds without it, rather than failing — a cascade can legitimately mix models that do and do not accept sampling parameters.

Response Objects

All response objects include usage metrics:

Advanced Usage

Automatic Fallback with LLMCascade

Use LLMCascade for automatic failover between providers:

from majordomo_llm import LLMCascade

# Providers are tried in order - first is primary, rest are fallbacks
cascade = LLMCascade([
    ("anthropic", "claude-sonnet-4-20250514"),  # Primary
    ("openai", "gpt-4.1"),                        # First fallback
    ("gemini", "gemini-2.5-flash"),              # Last resort
])

# If Anthropic fails, automatically tries OpenAI, then Gemini
response = await cascade.get_response("Hello!")

All response methods (get_response, get_json_response, get_structured_json_response, get_response_stream) support automatic fallback.

Optimal Routing (Majordomo Gateway)

The majordomo provider names a canonical open-weight model and lets the Majordomo gateway select the optimal backend (Fireworks, Together, …) at request time. Unlike LLMCascade (client-side failover on error), this is server-side provider selection — and the two compose.

It routes through the gateway, so base_url is required and MAJORDOMO_API_KEY must be set (auto-injected as the X-Majordomo-Key header):

import os
from majordomo_llm import get_llm_instance

llm = get_llm_instance(
    "majordomo", "glm-5.2",
    base_url=os.environ["MAJORDOMO_GATEWAY_URL"],
)

response = await llm.get_response("Hello!")

print(response.routed_provider)  # e.g. "fireworks" — the backend the gateway chose
print(response.routed_model)     # e.g. "accounts/fireworks/models/glm-5p2"
print(response.total_cost)       # priced from the routed backend's rates

Because the backend is known only after the call, cost is resolved from the gateway's X-Majordomo-Routed-Provider / X-Majordomo-Routed-Model response headers against that pair's rates in llm_config.yaml, rather than a fixed rate. An unconfigured routed pair degrades to 0.0 cost with a warning (token counts still stand). Text, JSON, structured, and streaming calls are all supported.

Direct Provider Access

You can also instantiate providers directly for more control:

from majordomo_llm import Anthropic

llm = Anthropic(
    model="claude-sonnet-4-20250514",
    input_cost=3.0,    # per million tokens
    output_cost=15.0,  # per million tokens
)

Web Search (Anthropic)

Enable web search for supported Claude models:

from majordomo_llm.providers.anthropic import Anthropic

llm = Anthropic(
    model="claude-sonnet-4-5-20250929",
    input_cost=3.0,
    output_cost=15.0,
    use_web_search=True,
)

Request Logging

Log all LLM requests asynchronously to a database with optional storage for request/response bodies. Logging is fire-and-forget and does not block your main request flow.

from majordomo_llm import get_llm_instance
from majordomo_llm.logging import LoggingLLM, PostgresAdapter, S3Adapter

async def main():
    # Create your LLM instance
    llm = get_llm_instance("anthropic", "claude-sonnet-4-20250514")

    # Set up database adapter (PostgreSQL, MySQL, or SQLite)
    db = await PostgresAdapter.create(
        host="localhost",
        port=5432,
        database="llm_logs",
        user="postgres",
        password="password",
    )

    # Optional: Set up S3 for storing request/response bodies
    storage = await S3Adapter.create(
        bucket="my-llm-logs",
        prefix="requests",  # optional, defaults to "llm-logs"
    )

    # Wrap your LLM with logging
    logged_llm = LoggingLLM(llm, db, storage)

    # Use as normal - all requests are logged automatically
    response = await logged_llm.get_response("Hello!")

    # Don't forget to close connections when done
    await logged_llm.close()

Local Development Setup

For local development and testing, use SQLite and local file storage:

from majordomo_llm import get_llm_instance
from majordomo_llm.logging import LoggingLLM, SqliteAdapter, FileStorageAdapter

async def main():
    llm = get_llm_instance("anthropic", "claude-sonnet-4-20250514")

    # SQLite for metrics (auto-creates database and table)
    db = await SqliteAdapter.create("llm_logs.db")

    # Local file storage for request/response bodies
    storage = await FileStorageAdapter.create("./request_logs")

    logged_llm = LoggingLLM(llm, db, storage)
    response = await logged_llm.get_response("Hello!")

    await logged_llm.close()

API Key Tracking

Track which API key was used for each request with optional human-readable aliases:

from majordomo_llm.providers.anthropic import Anthropic

# Create LLM with API key alias for attribution
llm = Anthropic(
    model="claude-sonnet-4-20250514",
    input_cost=3.0,
    output_cost=15.0,
    api_key_alias="production-team-1",  # Optional human-readable name
)

# The LoggingLLM wrapper automatically logs:
# - api_key_hash: First 16 chars of SHA256 hash (safe for logging)
# - api_key_alias: Your custom name (e.g., "production-team-1")

This is useful for:

• Tracking costs per team or application
• Debugging which key was used for specific requests
• Auditing API key usage patterns

Database Schema

Create the logging table using the included schema:

CREATE TABLE IF NOT EXISTS llm_requests (
    request_id VARCHAR(36) PRIMARY KEY,
    provider VARCHAR(50) NOT NULL,
    model VARCHAR(100) NOT NULL,
    timestamp TIMESTAMP NOT NULL,
    response_time FLOAT,
    input_tokens INTEGER,
    output_tokens INTEGER,
    cached_tokens INTEGER,
    input_cost DECIMAL(10, 8),
    output_cost DECIMAL(10, 8),
    total_cost DECIMAL(10, 8),
    s3_request_key VARCHAR(255),
    s3_response_key VARCHAR(255),
    status VARCHAR(20) NOT NULL,
    error_message TEXT,
    api_key_hash VARCHAR(16),
    api_key_alias VARCHAR(100)
);

Available Adapters

Database Adapters:

• PostgresAdapter - PostgreSQL via asyncpg
• MySQLAdapter - MySQL via aiomysql
• SqliteAdapter - SQLite via aiosqlite (great for local development)

Storage Adapters:

• S3Adapter - AWS S3 via aioboto3
• FileStorageAdapter - Local filesystem (great for local development)

Development

Setup

git clone https://github.com/superset-studio/majordomo-llm.git
cd majordomo-llm
uv sync --all-extras

Running Tests

uv run pytest

Type Checking

uv run mypy src/majordomo_llm

Linting

uv run ruff check src/majordomo_llm

Documentation

Build and preview the docs locally:

uv add --dev mkdocs mkdocs-material mkdocstrings[python] pymdown-extensions
uv run mkdocs serve

Pre-commit Hooks & Checks

Enable local checks (using uvx):

uvx pre-commit install
uvx pre-commit run --all-files

Hooks include private-key detection and basic hygiene checks. See .pre-commit-config.yaml.

Contributing

Contributions are welcome! Please see CONTRIBUTING.md for guidelines.

License

This project is licensed under the MIT License - see the LICENSE file for details.