Unit 4.4: AI Model Hub Integration

Introduction

You are adding an AI feature to TaskBoard: when a user submits a long task description, the API calls a large language model to produce a one-line summary, then stores that summary in PostgreSQL. AI Model Hub exposes hosted models behind an OpenAI-compatible REST API, so you do not run any GPUs and you do not manage model weights. You send a prompt to an inference endpoint, you get tokens back, and you pay per token.

This unit starts at the HTTP call. You will see the exact request and response shapes, wire them into Python with both requests and the OpenAI client, add the retry and timeout logic that production inference needs, and then connect the feature to the rest of TaskBoard through synchronous calls and a Kafka-queued asynchronous path. Every code block targets the real IONOS endpoint and real model identifiers, so you can copy, set your token, and run.

1. The Inference Endpoint and Authentication

AI Model Hub provides two API options: a native AI Model Hub API and an OpenAI-compatible API. The OpenAI-compatible API mirrors the OpenAI request and response structure, which lets you reuse existing OpenAI tooling and SDKs by changing only the base URL and the credentials. For application integration this is the path you want, because the request and response shapes are already familiar and the official OpenAI client libraries work unmodified.

The OpenAI-compatible base URL is https://openai.inference.de-txl.ionos.com/v1. Authentication uses a Bearer token, which is a JSON Web Token (JWT) with an expiration date. The AI Model Hub how-tos assume the token is held in the IONOS_API_TOKEN environment variable. Because the token is a JWT with an exp claim, a request that suddenly returns an authentication error usually means the token expired, not that the call is wrong. Check the exp claim and regenerate the token if it has passed.

1.1 First call with curl

Send a chat completion to confirm your token and connectivity before writing any application code. The endpoint is POST /v1/chat/completions and the request body carries a model identifier and a messages array.

export IONOS_API_TOKEN="your-jwt-token"

curl -s https://openai.inference.de-txl.ionos.com/v1/chat/completions \
  -H "Authorization: Bearer ${IONOS_API_TOKEN}" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "meta-llama/Llama-3.3-70B-Instruct",
    "messages": [
      {"role": "user", "content": "Summarize in one sentence: the deploy pipeline failed because the registry token expired."}
    ],
    "temperature": 0.2,
    "max_tokens": 60
  }'

The response is an OpenAI-style completion object. The generated text is in choices[0].message.content, and usage reports prompt_tokens, completion_tokens, and total_tokens, which you need for cost tracking.

{
  "id": "chatcmpl-123",
  "object": "chat.completion",
  "model": "meta-llama/Llama-3.3-70B-Instruct",
  "choices": [
    {
      "index": 0,
      "message": {"role": "assistant", "content": "The deploy pipeline failed because the container registry token had expired."},
      "finish_reason": "stop"
    }
  ],
  "usage": {"prompt_tokens": 28, "completion_tokens": 14, "total_tokens": 42}
}

1.2 Listing available models

The OpenAI-compatible API exposes a models endpoint to retrieve the list of available models and their details. Use it to discover the exact model identifier strings rather than hard-coding a guess, since the identifier is what the model field requires.

curl -s https://openai.inference.de-txl.ionos.com/v1/models \
  -H "Authorization: Bearer ${IONOS_API_TOKEN}" | python3 -m json.tool

The model identifier you pass must match the catalog exactly. For the chat models used in this unit the identifiers are meta-llama/Llama-3.3-70B-Instruct, openai/gpt-oss-120b, and mistralai/Mistral-Small-24B-Instruct. A typo in the identifier is one of the most common first-call failures.

2. Calling Model Hub from Application Code

In TaskBoard's API service you call inference from Python. You have two clean options: drive the HTTP API directly with requests, or use the official openai client pointed at the IONOS base URL. Both hit the same endpoint. The requests path keeps dependencies minimal; the OpenAI client gives you typed helpers, streaming iterators, and retry hooks for free.

2.1 Direct HTTP with requests

This is the lowest-dependency integration. It is also the clearest way to see exactly what crosses the wire, which helps when you debug.

import os
import requests

BASE_URL = "https://openai.inference.de-txl.ionos.com/v1"
TOKEN = os.environ["IONOS_API_TOKEN"]

def summarize(description: str) -> str:
    resp = requests.post(
        f"{BASE_URL}/chat/completions",
        headers={"Authorization": f"Bearer {TOKEN}"},
        json={
            "model": "meta-llama/Llama-3.3-70B-Instruct",
            "messages": [
                {"role": "system", "content": "You write one-sentence task summaries."},
                {"role": "user", "content": description},
            ],
            "temperature": 0.2,
            "max_tokens": 60,
        },
        timeout=30,
    )
    resp.raise_for_status()
    return resp.json()["choices"][0]["message"]["content"].strip()

Always set an explicit timeout. Inference latency varies with prompt length and model size, and a hung socket with no timeout will pin a worker thread in your API.

2.2 The OpenAI client pointed at IONOS

Because the API is OpenAI-compatible, the official openai Python package works when you override base_url and api_key. This is the most ergonomic option for application code.

import os
from openai import OpenAI

client = OpenAI(
    base_url="https://openai.inference.de-txl.ionos.com/v1",
    api_key=os.environ["IONOS_API_TOKEN"],
)

def summarize(description: str) -> str:
    completion = client.chat.completions.create(
        model="meta-llama/Llama-3.3-70B-Instruct",
        messages=[
            {"role": "system", "content": "You write one-sentence task summaries."},
            {"role": "user", "content": description},
        ],
        temperature=0.2,
        max_tokens=60,
    )
    return completion.choices[0].message.content.strip()

2.3 Streaming responses

For interactive UI you can stream tokens as they are generated by setting stream to true. When streaming, usage statistics are not included by default. To get usage in the final chunk you must explicitly set stream_options with include_usage enabled.

stream = client.chat.completions.create(
    model="openai/gpt-oss-120b",
    messages=[{"role": "user", "content": "Explain what a poison message is."}],
    stream=True,
    stream_options={"include_usage": True},
)
for chunk in stream:
    if chunk.choices and chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

The stream ends with a final data line containing usage followed by a [DONE] marker. If you skip stream_options, you lose the token counts you need for cost accounting.

3. Choosing a Model: Context Window and Price

Model selection is a code decision driven by two numbers: how many tokens the model accepts (context window) and what each million tokens costs. TaskBoard summaries are short, so the cheapest capable model is the right default; reserve larger models for tasks that genuinely need them.

The following table compares the chat models used in this unit. Prices are in EUR per million tokens.

As shown above, all three accept a 128000-token context window, so for a short summarization prompt the deciding factor is price. mistralai/Mistral-Small-24B-Instruct is the cheapest on both input and output and is a sound default for TaskBoard summaries. Move to a larger model only when summary quality on real task descriptions is demonstrably better and worth the per-token premium.

3.1 Cost-aware model configuration

Make the model a configuration value, not a literal scattered through the code, so you can switch by environment without redeploying logic.

import os

SUMMARY_MODEL = os.environ.get("SUMMARY_MODEL", "mistralai/Mistral-Small-24B-Instruct")

def estimate_cost_eur(prompt_tokens: int, completion_tokens: int,
                      in_price: float, out_price: float) -> float:
    return (prompt_tokens / 1_000_000) * in_price + (completion_tokens / 1_000_000) * out_price

print(estimate_cost_eur(28, 14, 0.10, 0.30))  # 28 prompt + 14 completion tokens, Mistral-Small pricing

3.2 Embeddings for semantic features

When TaskBoard needs semantic search over tasks, you generate vectors with an embedding model rather than a chat model. The endpoint is POST /v1/embeddings. The BAAI/bge-m3 model returns 1024-dimensional vectors and is priced at 0.02 EUR per million tokens.

def embed(text: str) -> list[float]:
    resp = client.embeddings.create(model="BAAI/bge-m3", input=text)
    return resp.data[0].embedding  # 1024 floats

vec = embed("Fix the expired registry token in the deploy pipeline")
assert len(vec) == 1024

Store the resulting vectors wherever your search layer lives. The embedding is deterministic for a given input and model, so identical text always produces the same vector, which makes embeddings a strong caching candidate.

4. Production Error Handling

Inference calls fail in ways your CRUD endpoints do not: the model is rate-limited, the request times out under load, or the response is well-formed JSON but the content is empty or off-format. Production code must handle all three. The most important one on IONOS is the rate limit.

4.1 Rate limits and 429 backoff

The general API rate limit is 5 requests per second (base) with a burst allowance of 10. When you exceed the limit, the service returns HTTP 429 Too Many Requests. Treat 429 as a signal to back off and retry, not as a hard failure. Use exponential backoff so a burst of TaskBoard activity does not hammer the endpoint.

import time
import requests

def chat_with_retry(payload: dict, max_attempts: int = 5) -> dict:
    delay = 1.0
    for attempt in range(max_attempts):
        resp = requests.post(
            f"{BASE_URL}/chat/completions",
            headers={"Authorization": f"Bearer {TOKEN}"},
            json=payload,
            timeout=30,
        )
        if resp.status_code == 429:
            retry_after = float(resp.headers.get("Retry-After", delay))
            time.sleep(retry_after)
            delay *= 2
            continue
        resp.raise_for_status()
        return resp.json()
    raise RuntimeError("inference rate-limited after retries")

Respect the Retry-After header when the service provides it, and fall back to your own exponential delay when it does not. Cap the number of attempts so a sustained outage surfaces as an error instead of an infinite loop.

4.2 Timeouts and response validation

A 200 response is not a guarantee of usable content. The model can return an empty string or text that breaks your downstream assumptions. Validate before you store.

def safe_summary(description: str) -> str:
    payload = {
        "model": SUMMARY_MODEL,
        "messages": [
            {"role": "system", "content": "Reply with exactly one sentence."},
            {"role": "user", "content": description},
        ],
        "max_tokens": 60,
    }
    data = chat_with_retry(payload)
    choices = data.get("choices") or []
    if not choices:
        raise ValueError("no choices in inference response")
    content = (choices[0]["message"]["content"] or "").strip()
    if not content:
        raise ValueError("empty summary returned")
    return content

For structured extraction, the chat completions API supports a response_format with a JSON schema and strict enabled, which constrains the model to emit valid JSON matching your schema. That is the reliable way to get machine-parseable output instead of free text you have to parse defensively.

5. Integration Patterns and Cost Control

You can call inference two ways inside TaskBoard. Synchronous inference runs inside the request that needs the result. Asynchronous inference queues the work and processes it out of band. The right choice depends on whether the user is waiting for the answer.

5.1 Synchronous: summarize on create

When the user submits a task and expects the summary back immediately, call inference inside the request handler and write the summary to PostgreSQL in the same transaction.

def create_task(db, description: str) -> int:
    summary = safe_summary(description)
    row = db.execute(
        "INSERT INTO tasks (description, summary) VALUES (%s, %s) RETURNING id",
        (description, summary),
    ).fetchone()
    db.commit()
    return row[0]

Keep the synchronous path behind a timeout and a retry budget so a slow model does not stall the user-facing request indefinitely.

5.2 Asynchronous: queue inference through Kafka

When the summary is not needed instantly, decouple it. The API writes the task immediately and produces a task-change event to Kafka (see Unit 4.3); a worker consumes the event, calls inference, and updates the row. This keeps the create request fast and isolates inference latency and rate limits from the user path.

def handle_event(event: dict, db):  # worker side: consume task event, summarize, persist
    task_id = event["task_id"]
    description = event["description"]
    try:
        summary = safe_summary(description)
    except (ValueError, RuntimeError):
        # route to a dead-letter topic; do not block the consumer group
        produce_dead_letter(event)
        return
    db.execute("UPDATE tasks SET summary = %s WHERE id = %s", (summary, task_id))
    db.commit()

The asynchronous pattern also smooths bursts: the 5 requests-per-second limit is far easier to honor when a single consumer drains a queue at a controlled rate than when many web workers call inference concurrently.

5.3 Caching results in Redis

Inference is the most expensive call in this feature, so do not pay for it twice. Cache by a hash of the model plus the input. Identical descriptions then cost nothing after the first call. This pairs naturally with TaskBoard's existing In-Memory DB (Redis) cache from Unit 4.1.

import hashlib
import redis

r = redis.Redis(host="taskboard-redis", port=6379, ssl=True)

def cached_summary(description: str) -> str:
    key = "sum:" + hashlib.sha256(
        (SUMMARY_MODEL + "|" + description).encode()
    ).hexdigest()
    hit = r.get(key)
    if hit:
        return hit.decode()
    summary = safe_summary(description)
    r.set(key, summary, ex=86400)  # 24h TTL
    return summary

Because inference is stateless and prompts and outputs are discarded at the end of each session, the only durable record of a result is the one you choose to keep. Caching is therefore both a cost lever and your store of computed results.

5.4 Data residency and privacy

AI Model Hub operates as a stateless service: prompts and outputs are discarded at the end of each session, are not logged or recorded, and are not reused for model training. Customer data is not used for training under any circumstance. All AI Model Hub services, including model inference endpoints and managed vector databases, are hosted in ISO 27001-certified data centers in Germany, and processing is fully GDPR-compliant. For TaskBoard this means task descriptions sent for summarization stay within the German data center boundary and never enter a training set.

5.5 Read-only agentic access with the IONOS CLOUD MCP Server

When the integration you need is not "call a model" but "let an AI agent inspect your IONOS infrastructure," the IONOS CLOUD MCP Server is the developer-facing path. It is a local binary that implements the Model Context Protocol (MCP), speaking JSON-RPC over stdio to an AI client or autonomous agent. It gives that client read-only access to 100+ tools across six products: Compute Engine, Object Storage, Cloud DNS, Certificate Manager, Billing, and Activity Log. The tools are inspection-only by design, so the binary cannot create, update, or delete any resource.

The MCP Server is an alternative to wiring your own direct API or SDK calls for agentic and automation read scenarios. Instead of writing client code against each product API, you run the binary as a local subprocess; the AI client discovers the tools and calls them in an agentic loop, and the binary calls the IONOS Cloud API directly over HTTPS using your API token. For fully sovereign workflows it pairs with AI Model Hub, so both the inference step and the infrastructure-inspection step stay within the IONOS perimeter. Treat tool outputs as data that leaves that perimeter once the AI client reads them.

API Reference Quick Card

Key API endpoints for AI Model Hub integration:

Base URL (OpenAI-compatible): https://openai.inference.de-txl.ionos.com/v1 Native API docs: https://api.ionos.com/docs/inference-modelhub/v1 Authentication: Authorization: Bearer <IONOS_API_TOKEN> (JWT with exp claim)

Code Lab

Objective: Add AI summarization to TaskBoard. Call AI Model Hub to summarize a task description, handle a 429 with backoff, and cache the result in Redis so a repeat summary costs no tokens.

Prerequisites:

• IONOS Cloud account with an API token (JWT) in IONOS_API_TOKEN
• Python 3.10+ with pip install openai requests redis
• A reachable Redis instance (or local redis for the lab)

Step 1: Export your token and confirm connectivity

export IONOS_API_TOKEN="your-jwt-token"
curl -s https://openai.inference.de-txl.ionos.com/v1/models \
  -H "Authorization: Bearer ${IONOS_API_TOKEN}" | python3 -m json.tool | head

Expected output:

{
    "object": "list",
    "data": [
        { "id": "meta-llama/Llama-3.3-70B-Instruct", ... }

Step 2: Make a first summary call

from openai import OpenAI
import os
client = OpenAI(base_url="https://openai.inference.de-txl.ionos.com/v1",
                api_key=os.environ["IONOS_API_TOKEN"])
c = client.chat.completions.create(
    model="mistralai/Mistral-Small-24B-Instruct",
    messages=[{"role": "user", "content": "Summarize in one sentence: registry token expired so the deploy failed."}],
    max_tokens=60, temperature=0.2)
print(c.choices[0].message.content)
print("tokens:", c.usage.total_tokens)

Expected output:

The deployment failed because the container registry token had expired.
tokens: 41

Step 3: Add retry-on-429 with exponential backoff

import time, requests, os
BASE="https://openai.inference.de-txl.ionos.com/v1"
def call(payload, attempts=5):
    delay=1.0
    for _ in range(attempts):
        r=requests.post(f"{BASE}/chat/completions",
            headers={"Authorization":f"Bearer {os.environ['IONOS_API_TOKEN']}"},
            json=payload, timeout=30)
        if r.status_code==429:
            time.sleep(float(r.headers.get("Retry-After", delay))); delay*=2; continue
        r.raise_for_status(); return r.json()
    raise RuntimeError("rate-limited")

Expected output: no error on a normal call; a 429 triggers a wait and retry instead of a crash.

Step 4: Validate the response before using it

def summary(text):
    data=call({"model":"mistralai/Mistral-Small-24B-Instruct",
               "messages":[{"role":"user","content":text}],"max_tokens":60})
    out=(data["choices"][0]["message"]["content"] or "").strip()
    if not out: raise ValueError("empty summary")
    return out
print(summary("The CI job failed on the embeddings step."))

Expected output:

The CI job failed during the embeddings step.

Step 5: Cache results in Redis

import hashlib, redis
r=redis.Redis(host="localhost", port=6379)
MODEL="mistralai/Mistral-Small-24B-Instruct"
def cached(text):
    k="sum:"+hashlib.sha256((MODEL+"|"+text).encode()).hexdigest()
    hit=r.get(k)
    if hit: return hit.decode()
    s=summary(text); r.set(k, s, ex=86400); return s

Step 6: Verify the cache eliminates the second call's cost

import time
t=time.time(); cached("Same description here."); print("miss", round(time.time()-t,2))
t=time.time(); cached("Same description here."); print("hit ", round(time.time()-t,2))

Expected output:

miss 0.9
hit  0.0

Step 7: Generate an embedding for semantic search

e=client.embeddings.create(model="BAAI/bge-m3", input="Fix the expired registry token")
print(len(e.data[0].embedding))

Expected output:

1024

Validation Checklist:

• [ ] /v1/models returns a model list with your token
• [ ] A chat completion returns content and a usage token count
• [ ] A 429 triggers backoff and retry rather than an exception
• [ ] A repeated summary is served from Redis with no token cost
• [ ] BAAI/bge-m3 returns a 1024-dimension vector

Cleanup:

redis-cli --scan --pattern 'sum:*' | xargs -r redis-cli del

Common Pitfalls

Developer mistakes to avoid with AI Model Hub integration:

1. Treating a 429 as a fatal error

   • Problem: A burst of TaskBoard activity returns HTTP 429 Too Many Requests and your handler raises, failing user requests.
   • Why it happens: The general API rate limit is 5 requests per second (base) with a burst of 10; concurrent web workers exceed it easily.
   • Fix: Catch 429, honor Retry-After, and retry with exponential backoff. Push inference onto a Kafka-drained worker to control concurrency.

2. Expired JWT mistaken for a code bug

   • Problem: Calls that worked yesterday return an authentication error today, and you start debugging headers and payloads.
   • Why it happens: The Bearer token is a JWT with an exp claim and a fixed expiration date. When it passes, every call fails authentication.
   • Fix: Decode the token's exp claim and regenerate the token when expired. Treat auth failures as a token-lifecycle check first, not a request-format bug.

3. Calling inference synchronously and paying for duplicates

   • Problem: Latency spikes on task creation and the token bill climbs because identical descriptions are summarized repeatedly.
   • Why it happens: Inference runs inside the request path with no caching, so the same input costs tokens every time.
   • Fix: Cache by a hash of model plus input in Redis with a TTL, and move non-urgent summaries to an asynchronous Kafka worker so the create request stays fast.

Summary

You can now integrate AI Model Hub into application code. You call the OpenAI-compatible endpoint at https://openai.inference.de-txl.ionos.com/v1 with a Bearer JWT, drive it from requests or the official OpenAI client, stream when the UI needs it, and generate embeddings with BAAI/bge-m3. You select models by context window and per-token price, and you wrap every call in timeout, 429 backoff, and response validation. You wired the feature into TaskBoard both synchronously and through a Kafka-queued worker, and you cache results in Redis so repeated inputs cost nothing.

Key Points:

• The OpenAI-compatible API at /v1/chat/completions, /v1/embeddings, /v1/models, and /v1/images/generations lets the official OpenAI client work by overriding base_url and api_key
• Authentication is a Bearer JWT in IONOS_API_TOKEN; an expired exp claim is the most common cause of sudden auth failures
• The general rate limit is 5 requests per second base, 10 burst; exceeding it returns HTTP 429, which you handle with Retry-After and exponential backoff
• Model choice is a cost decision: all three chat models share a 128000-token context window, so price separates them, with mistralai/Mistral-Small-24B-Instruct the cheapest
• The service is stateless and German-hosted; prompts and outputs are discarded per session and never used for training, so Redis caching is both your cost lever and your only durable store of results
• The IONOS CLOUD MCP Server is a local binary giving AI agents read-only access to 100+ tools across Compute Engine, Object Storage, Cloud DNS, Certificate Manager, Billing, and Activity Log over MCP (JSON-RPC on stdio); it is an alternative to direct API or SDK calls for agentic read scenarios and pairs with AI Model Hub for sovereign workflows

Important Terminology:

• OpenAI-compatible API: An endpoint that mirrors OpenAI's request and response structure, letting OpenAI SDKs work against IONOS by changing the base URL and key
• Context window: The maximum number of tokens (prompt plus completion) a model accepts in one request; 128000 for the chat models in this unit
• Token-based pricing: Cost charged per million input and output tokens; tracked via the usage object in each response
• Exponential backoff: A retry strategy that doubles the wait between attempts after a 429, preventing a thundering-herd retry storm
• Stateless inference: Each request stands alone; AI Model Hub discards prompts and outputs at session end and does not log, store, or train on them
• IONOS CLOUD MCP Server: A local binary implementing the Model Context Protocol (JSON-RPC over stdio) that exposes read-only inspection tools across six IONOS products to an AI client, calling the IONOS Cloud API directly without any third-party AI provider in the data path

Next Steps

Continue Learning: Unit 4.5: Knowledge Check - Service Integration

Related Topics:

• Unit 4.1: Database and Caching Integration
• Unit 4.3: Event Streaming Integration