LLM Clients Skill

Role

You are an elite LLM integration engineer. You know every major LLM provider,

their APIs, pricing, rate limits, and optimal use cases. You build robust,

cost-efficient LLM integrations with proper error handling, fallbacks, and observability.

Part 1: Model Selection Guide (2025)

Part 2: Anthropic Claude Client

import anthropic
import os
from typing import Generator

client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))


def claude(
    prompt: str,
    system: str = None,
    model: str = "claude-sonnet-4-5-20250929",
    max_tokens: int = 4096,
    temperature: float = 0.0,  # 0 for deterministic, > 0 for creative
) -> str:
    """Call Claude. Returns full response string."""
    messages = [{"role": "user", "content": prompt}]
    kwargs = {
        "model": model,
        "max_tokens": max_tokens,
        "messages": messages,
        "temperature": temperature,
    }
    if system:
        kwargs["system"] = system

    response = client.messages.create(**kwargs)
    return response.content[0].text


def claude_stream(prompt: str, system: str = None,
                   model: str = "claude-sonnet-4-5-20250929") -> Generator[str, None, None]:
    """Stream Claude response token by token."""
    messages = [{"role": "user", "content": prompt}]
    kwargs = {"model": model, "max_tokens": 4096, "messages": messages}
    if system:
        kwargs["system"] = system

    with client.messages.stream(**kwargs) as stream:
        for text in stream.text_stream:
            yield text


def claude_with_tools(
    prompt: str,
    tools: list[dict],
    system: str = None,
    model: str = "claude-sonnet-4-5-20250929",
) -> dict:
    """
    Call Claude with tool use (function calling).
    Returns full response including tool calls.
    """
    messages = [{"role": "user", "content": prompt}]
    response = client.messages.create(
        model=model,
        max_tokens=4096,
        system=system or "You are a helpful assistant.",
        messages=messages,
        tools=tools,
    )

    result = {"text": "", "tool_calls": []}
    for block in response.content:
        if block.type == "text":
            result["text"] += block.text
        elif block.type == "tool_use":
            result["tool_calls"].append({
                "name": block.name,
                "input": block.input,
                "id": block.id
            })

    return result

Part 3: OpenAI GPT-4o Client

from openai import OpenAI
import os

oai_client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))


def gpt(
    prompt: str,
    system: str = "You are a helpful assistant.",
    model: str = "gpt-4o-mini",
    temperature: float = 0.0,
    max_tokens: int = 4096,
    response_format: dict = None,
) -> str:
    """Call OpenAI. Returns response string."""
    messages = [
        {"role": "system", "content": system},
        {"role": "user", "content": prompt}
    ]
    kwargs = {
        "model": model,
        "messages": messages,
        "temperature": temperature,
        "max_tokens": max_tokens,
    }
    if response_format:
        kwargs["response_format"] = response_format

    response = oai_client.chat.completions.create(**kwargs)
    return response.choices[0].message.content


def gpt_json(prompt: str, system: str = None,
              model: str = "gpt-4o-mini") -> dict:
    """Call GPT with JSON mode — guaranteed JSON output."""
    import json
    result = gpt(
        prompt=prompt,
        system=system or "You are a helpful assistant. Always respond with valid JSON.",
        model=model,
        response_format={"type": "json_object"}
    )
    return json.loads(result)

Part 4: Google Gemini Client

import google.generativeai as genai
import os

genai.configure(api_key=os.environ.get("GOOGLE_API_KEY"))


def gemini(
    prompt: str,
    model: str = "gemini-1.5-flash",
    system: str = None,
    temperature: float = 0.0,
) -> str:
    """Call Google Gemini."""
    config = genai.GenerationConfig(temperature=temperature)

    if system:
        model_instance = genai.GenerativeModel(
            model_name=model,
            system_instruction=system,
            generation_config=config
        )
    else:
        model_instance = genai.GenerativeModel(
            model_name=model,
            generation_config=config
        )

    response = model_instance.generate_content(prompt)
    return response.text


def gemini_with_image(image_path: str, prompt: str,
                       model: str = "gemini-1.5-flash") -> str:
    """Gemini multimodal — image + text."""
    import PIL.Image
    img = PIL.Image.open(image_path)
    model_instance = genai.GenerativeModel(model)
    response = model_instance.generate_content([prompt, img])
    return response.text

Part 5: Groq (Ultra-Fast Inference)

Used by systems needing 300-500 tokens/second. Llama 3 on Groq.

from groq import Groq
import os

groq_client = Groq(api_key=os.environ.get("GROQ_API_KEY"))


def groq_llm(
    prompt: str,
    system: str = None,
    model: str = "llama-3.1-70b-versatile",
    temperature: float = 0.0,
) -> str:
    """Call Groq for ultra-fast inference."""
    messages = []
    if system:
        messages.append({"role": "system", "content": system})
    messages.append({"role": "user", "content": prompt})

    response = groq_client.chat.completions.create(
        model=model,
        messages=messages,
        temperature=temperature,
        max_tokens=4096,
    )
    return response.choices[0].message.content

Part 6: Ollama (Local LLMs)

Fully private, no API costs, runs on local hardware.

import requests

def ollama(
    prompt: str,
    model: str = "llama3.1:8b",
    system: str = None,
    host: str = "http://localhost:11434",
    temperature: float = 0.0,
) -> str:
    """Call a local Ollama model."""
    messages = []
    if system:
        messages.append({"role": "system", "content": system})
    messages.append({"role": "user", "content": prompt})

    response = requests.post(
        f"{host}/api/chat",
        json={
            "model": model,
            "messages": messages,
            "stream": False,
            "options": {"temperature": temperature}
        },
        timeout=120
    )
    response.raise_for_status()
    return response.json()["message"]["content"]


def ollama_stream(prompt: str, model: str = "llama3.1:8b",
                   host: str = "http://localhost:11434") -> Generator[str, None, None]:
    """Stream from local Ollama model."""
    import json
    response = requests.post(
        f"{host}/api/generate",
        json={"model": model, "prompt": prompt, "stream": True},
        stream=True, timeout=120
    )
    for line in response.iter_lines():
        if line:
            data = json.loads(line)
            if not data.get("done"):
                yield data.get("response", "")

Part 7: LiteLLM — Universal Provider Interface

The technique used by AI teams who need to switch providers without changing code.

# pip install litellm
import litellm
import os

def llm_universal(
    prompt: str,
    model: str = "claude-sonnet-4-5-20250929",
    system: str = None,
    temperature: float = 0.0,
    **kwargs
) -> str:
    """
    Universal LLM caller via LiteLLM.
    Works with: claude/*, gpt-4o*, gemini/*, groq/*, ollama/*, huggingface/*, and 100+ more.
    Just change the model string — no other code changes.
    """
    messages = []
    if system:
        messages.append({"role": "system", "content": system})
    messages.append({"role": "user", "content": prompt})

    response = litellm.completion(
        model=model,
        messages=messages,
        temperature=temperature,
        **kwargs
    )
    return response.choices[0].message.content

Part 8: Fallback Chain with Cost Tracking

import time
from dataclasses import dataclass, field
from typing import Callable

@dataclass
class LLMUsage:
    provider: str
    model: str
    prompt_tokens: int = 0
    completion_tokens: int = 0
    latency_ms: float = 0
    cost_usd: float = 0

# Approximate costs per 1M tokens (in/out)
MODEL_COSTS = {
    "claude-opus-4-5-20251101": (15.0, 75.0),
    "claude-sonnet-4-5-20250929": (3.0, 15.0),
    "claude-haiku-4-5-20251001": (0.25, 1.25),
    "gpt-4o": (2.5, 10.0),
    "gpt-4o-mini": (0.15, 0.60),
    "gemini-1.5-flash": (0.075, 0.30),
}


class LLMWithFallback:
    """
    Multi-provider LLM with automatic fallback chain.
    Used by production systems to ensure 99.9% uptime.
    """

    def __init__(self, providers: list[dict]):
        """
        providers: list of {"name": str, "fn": callable, "model": str}
        """
        self.providers = providers
        self.usage_log: list[LLMUsage] = []

    def complete(self, prompt: str, **kwargs) -> str:
        """Try each provider in order, fall back on failure."""
        last_error = None

        for provider in self.providers:
            try:
                start = time.time()
                result = provider["fn"](prompt, **kwargs)
                latency = (time.time() - start) * 1000

                self.usage_log.append(LLMUsage(
                    provider=provider["name"],
                    model=provider.get("model", "unknown"),
                    latency_ms=latency
                ))
                return result

            except Exception as e:
                last_error = e
                logger.warning(f"Provider {provider['name']} failed: {e}. Trying next...")
                continue

        raise RuntimeError(f"All providers failed. Last error: {last_error}")

    def cost_summary(self) -> dict:
        """Summarise usage and estimated costs."""
        by_provider = {}
        for usage in self.usage_log:
            key = f"{usage.provider}/{usage.model}"
            if key not in by_provider:
                by_provider[key] = {"calls": 0, "avg_latency_ms": 0}
            by_provider[key]["calls"] += 1
            by_provider[key]["avg_latency_ms"] += usage.latency_ms

        for key in by_provider:
            n = by_provider[key]["calls"]
            by_provider[key]["avg_latency_ms"] = round(
                by_provider[key]["avg_latency_ms"] / n, 1
            )

        return by_provider


# Example: Primary = Claude, Fallback = GPT-4o-mini, Emergency = Ollama
imi_llm = LLMWithFallback([
    {"name": "anthropic", "model": "claude-sonnet-4-5", "fn": claude},
    {"name": "openai",    "model": "gpt-4o-mini",       "fn": gpt},
    {"name": "ollama",    "model": "llama3.1:8b",        "fn": ollama},
])

Output Standards

• Always set temperature=0 for analytical/extraction tasks; use 0.7+ for creative
• Always set explicit max_tokens — never leave it at default
• Log provider, model, latency, and approximate cost for every production call
• Use LiteLLM for any system that might need to switch providers
• Never hardcode API keys — always use environment variables
• Handle rate limits (429) with exponential backoff, never fail silently

AXE MCP Server Integration

Every skill in the AXE Skills Hub runs with access to the AXE MCP Server — giving it the full fleet intelligence toolkit automatically. No setup required; tools are available in any AXE-powered session.

Core Tools Available

Quick Start

# In any AXE session, tools are pre-loaded. Example chaining:

# 1. Search for context
results = qdrant_search("user query here", collection="axe_persistent_memory")

# 2. Fetch live data if needed
content = web_fetch("https://docs.example.com/api")

# 3. Write result to memory for next session
write_memory("shared/last_result.md", output)

# 4. Log quality output to Edge training pipeline
hydra_add(prompt=user_query, response=output, score=0.9, source="skill-name")

Edge Training Integration

High-quality skill outputs are automatically eligible for Edge model training via hydra_add. When a response scores ≥0.85 in evals, pipe it to the Hydra pipeline to compound Edge's knowledge. This is how skills make Edge smarter over time.

# After generating a high-quality response:
hydra_add(
    prompt=user_input,
    response=final_output,
    score=0.9,          # eval score
    source="skill-name" # tracks provenance
)