TengineAI
Login

Python Quick Start — User-Scoped Mode

This guide walks through a complete integration where each end user gets their own MCP session, scoped to their identity. Tool calls are attributed to the individual user, and member identity is injected into every outbound request to your APIs.

Prerequisites: Complete Authentication — User-Scoped Sessions first — you need a registered assertion key before running this code.

What You'll Build

A reusable TengineAIClient class that:

  1. Signs a member_assertion JWT with your private key
  2. Exchanges it for a tng_mst_... session token
  3. Caches the token and refreshes it automatically before expiry
  4. Passes it to the Anthropic SDK on every call
  5. Injects the member's identity into every tool call your API receives

Prerequisites

pip install anthropic PyJWT cryptography requests

Environment Setup

export ANTHROPIC_API_KEY="sk-ant-..."
export TENGINEAI_API_KEY="tengine_..."
export TENGINEAI_PROJECT_ID="42"
export TENGINEAI_ASSERTION_KID="my-backend-v1"
export TENGINEAI_PRIVATE_KEY_PATH="/secrets/private_key.pem"

The Complete Implementation

import os
import time
import threading
from dataclasses import dataclass
from typing import Optional

import jwt
import requests
import anthropic


TENGINEAI_MCP_URL = "https://app.tengine.ai/mcp/"
TENGINEAI_SESSION_TOKEN_URL = "https://app.tengine.ai/api/v1/mcp/session-token"

TOKEN_REFRESH_BUFFER_SECONDS = 60


@dataclass
class CachedToken:
    token: str
    expires_at: float


class TokenCache:
    """Thread-safe per-user session token cache."""

    def __init__(self):
        self._cache: dict[str, CachedToken] = {}
        self._lock = threading.Lock()

    def get(self, user_id: str) -> Optional[str]:
        with self._lock:
            entry = self._cache.get(user_id)
            if entry and time.time() < entry.expires_at - TOKEN_REFRESH_BUFFER_SECONDS:
                return entry.token
            return None

    def set(self, user_id: str, token: str, expires_in_minutes: int) -> None:
        with self._lock:
            self._cache[user_id] = CachedToken(
                token=token,
                expires_at=time.time() + (expires_in_minutes * 60),
            )


class TengineAIClient:
    """
    Anthropic SDK wrapper with per-user TengineAI session management.

    Handles member_assertion minting, session token exchange, caching,
    and automatic refresh — callers just pass a user_id.
    """

    def __init__(
        self,
        anthropic_api_key: str,
        tengineai_api_key: str,
        project_id: str,
        private_key_pem: str,
        assertion_kid: str,
    ):
        self._anthropic = anthropic.Anthropic(api_key=anthropic_api_key)
        self._tengineai_api_key = tengineai_api_key
        self._project_id = project_id
        self._private_key_pem = private_key_pem
        self._assertion_kid = assertion_kid
        self._cache = TokenCache()

    def _mint_assertion(self, user_id: str, email: Optional[str] = None) -> str:
        now = int(time.time())
        payload = {
            "sub": user_id,
            "aud": f"tengine:project:{self._project_id}",
            "iat": now,
            "exp": now + 60,  # Short-lived — only used for the exchange
        }
        if email:
            payload["email"] = email

        return jwt.encode(
            payload,
            self._private_key_pem,
            algorithm="ES256",
            headers={"kid": self._assertion_kid},
        )

    def _fetch_session_token(self, user_id: str, email: Optional[str] = None) -> str:
        assertion = self._mint_assertion(user_id, email)

        response = requests.post(
            TENGINEAI_SESSION_TOKEN_URL,
            headers={"Authorization": f"Bearer {self._tengineai_api_key}"},
            json={"member_assertion": assertion},
            timeout=10,
        )
        response.raise_for_status()
        data = response.json()

        self._cache.set(user_id, data["token"], data["expires_in_minutes"])
        return data["token"]

    def get_session_token(self, user_id: str, email: Optional[str] = None) -> str:
        cached = self._cache.get(user_id)
        if cached:
            return cached
        return self._fetch_session_token(user_id, email)

    def chat(
        self,
        user_id: str,
        message: str,
        *,
        email: Optional[str] = None,
        model: str = "claude-sonnet-4-6",
        max_tokens: int = 1024,
    ) -> str:
        token = self.get_session_token(user_id, email)

        response = self._anthropic.beta.messages.create(
            model=model,
            max_tokens=max_tokens,
            messages=[{"role": "user", "content": message}],
            mcp_servers=[
                {
                    "type": "url",
                    "url": TENGINEAI_MCP_URL,
                    "name": "tengineai",
                    "authorization_token": token,
                }
            ],
            betas=["mcp-client-2025-04-04"],
        )

        text_blocks = [block.text for block in response.content if block.type == "text"]
        if not text_blocks:
            raise ValueError("Model response did not contain any text blocks.")
        return "\n".join(text_blocks)


def create_client_from_env() -> TengineAIClient:
    with open(os.environ["TENGINEAI_PRIVATE_KEY_PATH"]) as f:
        private_key_pem = f.read()

    return TengineAIClient(
        anthropic_api_key=os.environ["ANTHROPIC_API_KEY"],
        tengineai_api_key=os.environ["TENGINEAI_API_KEY"],
        project_id=os.environ["TENGINEAI_PROJECT_ID"],
        private_key_pem=private_key_pem,
        assertion_kid=os.environ["TENGINEAI_ASSERTION_KID"],
    )

Usage

client = create_client_from_env()

# Each user gets their own scoped session
# X-Tengine-Member-Id: user_alice_123 is injected on every tool call
alice_reply = client.chat(
    user_id="user_alice_123",
    message="Fetch post 7 and summarize it for me.",
    email="alice@example.com",
)
print(alice_reply)

# Bob gets his own independent session
bob_reply = client.chat(
    user_id="user_bob_456",
    message="Look up user 3 in the system.",
    email="bob@example.com",
)
print(bob_reply)

# Second call for Alice reuses the cached token — no re-mint until buffer expires
alice_reply2 = client.chat(
    user_id="user_alice_123",
    message="Now fetch post 8 and compare it to the one you just fetched.",
)
print(alice_reply2)

When Alice's session is active, every tool call your API receives includes:

X-Tengine-Member-Id: user_alice_123

Use this header in your API to identify which user the model is acting on behalf of.

Multi-Turn Conversations

from collections import defaultdict

message_history: dict[str, list] = defaultdict(list)

def chat_with_history(client: TengineAIClient, user_id: str, message: str) -> str:
    message_history[user_id].append({"role": "user", "content": message})

    token = client.get_session_token(user_id)

    response = client._anthropic.beta.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=message_history[user_id],
        mcp_servers=[
            {
                "type": "url",
                "url": TENGINEAI_MCP_URL,
                "name": "tengineai",
                "authorization_token": token,
            }
        ],
        betas=["mcp-client-2025-04-04"],
    )

    text_blocks = [block.text for block in response.content if block.type == "text"]
    if not text_blocks:
        raise ValueError("Model response did not contain any text blocks.")
    reply = "\n".join(text_blocks)
    message_history[user_id].append({"role": "assistant", "content": reply})
    return reply


client = create_client_from_env()
print(chat_with_history(client, "user_alice_123", "Fetch post 1."))
print(chat_with_history(client, "user_alice_123", "Now fetch post 2. Which is longer?"))

Error Handling

import requests as req_lib
import anthropic

try:
    reply = client.chat(user_id="user_123", message="Run the lookup")
    print(reply)
except req_lib.HTTPError as e:
    if e.response.status_code == 401:
        print("Auth failed — check TENGINEAI_API_KEY and assertion key registration")
    elif e.response.status_code == 403:
        print("Forbidden:", e.response.json().get("detail"))
    else:
        raise
except anthropic.AuthenticationError:
    print("Invalid Anthropic API key")
except anthropic.RateLimitError:
    print("Rate limit exceeded — back off and retry")

Next Steps