SentinelMeshAIai SDK

Official client libraries for Python and JavaScript/TypeScript. Full API coverage with typed methods, async streaming, auto-retry, and zero heavy dependencies.

Supported Languages

Use Cases

• Chatbots & assistants — multi-turn conversations with persistent memory sessions
• Code tools — route code prompts to Codestral, review PRs, generate unit tests
• RAG pipelines — integrate Command-R-Plus for retrieval-augmented generation over your documents
• Data extraction — structured JSON output from unstructured text or images
• Cost-optimised batch processing — cache semantically similar requests, use DeepSeek for bulk tasks
• Enterprise automation — webhook-driven workflows, event triggers, 12 live integration connectors

Python Installation

bashCopypip install sentinelmeshai

bashCopypip install "sentinelmeshai[dev]"
# Installs: pytest, pytest-asyncio, respx

Node.js / TypeScript Installation

bashCopynpm install sentinelmeshai

bashCopyyarn add sentinelmeshai

Authentication Setup

Set your API key as an environment variable. The SDK reads it automatically.

pythonCopyimport sentinelmeshai as sm

# Option A: read from environment (recommended)
sm.api_key = os.getenv("SENTINELMESHAIAI_API_KEY")

# Option B: set directly (fine for scripts, never for services)
sm.api_key = "sm_live_..."

# Option C: per-client (multiple accounts)
from sentinelmeshai import SentinelMeshAIai
client = SentinelMeshAIAI(
    api_key  = os.getenv("SENTINELMESHAIAI_API_KEY"),
    base_url = "https://api.sentinelmeshai.com",
    timeout  = 60.0,
)

javascriptCopyimport SentinelMeshAIai from 'sentinelmeshai';

// Option A: auto-reads SENTINELMESHAIAI_API_KEY from process.env
const sm = new SentinelMeshAI();

// Option B: explicit
const sm = new SentinelMeshAI({
  apiKey:  process.env.SENTINELMESHAIAI_API_KEY,
  baseUrl: 'https://api.sentinelmeshai.com',
  timeout: 60_000,
});

bashCopy# .env (add to .gitignore — never commit this file)
SENTINELMESHAI_API_KEY=sm_live_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx

# Python: load with python-dotenv
# from dotenv import load_dotenv; load_dotenv()

# Node.js: load with dotenv
# require('dotenv').config()  OR  import 'dotenv/config'

Python Quickstart

pythonCopyimport os, sentinelmeshai as sm

sm.api_key = os.getenv("SENTINELMESHAIAI_API_KEY")

# Synchronous — simplest usage
result = sm.run("Summarise the key points of the Kyoto Protocol.")
print(result["output"])
print(f"Cost: ${result['cost_usd']:.6f} | Latency: {result['latency_ms']}ms")

# Async — for Django async views, FastAPI, asyncio scripts
import asyncio
async def main():
    result = await sm.arun("Explain backpropagation in simple terms.")
    print(result["output"])

asyncio.run(main())

SentinelMeshAI Class (Python)

run() — Synchronous

pythonCopyresult = sm.run(
    "Write a Python function to detect palindromes.",
    strategy    = "code_specialist",
    temperature = 0.1,
    max_tokens  = 2048,
    metadata    = {"user_id": "u_12345", "feature": "code_gen"},
)
print(result["output"])
# → "def is_palindrome(s: str) -> bool:\n    s = s.lower()..."

arun() — Async

pythonCopy# FastAPI endpoint example
from fastapi import FastAPI
import sentinelmeshai as sm, os

sm.api_key = os.getenv("SENTINELMESHAI_API_KEY")
app = FastAPI()

@app.post("/ask")
async def ask(prompt: str):
    result = await sm.arun(prompt, strategy="standard")
    return {"answer": result["output"], "cost": result["cost_usd"]}

astream() — Async Streaming

pythonCopyimport asyncio, sentinelmeshai as sm, os

sm.api_key = os.getenv("SENTINELMESHAI_API_KEY")

async def stream_answer():
    print("Answer: ", end="", flush=True)
    async for chunk in sm.astream("Write a haiku about distributed systems."):
        print(chunk, end="", flush=True)
    print()  # newline at end

asyncio.run(stream_answer())

# Django async view with streaming response
from django.http import StreamingHttpResponse
async def streaming_view(request):
    async def generator():
        async for chunk in sm.astream(request.GET["q"]):
            yield chunk.encode()
    return StreamingHttpResponse(generator(), content_type="text/plain")

run_with_memory() — Session Memory

pythonCopyimport uuid, sentinelmeshai as sm

session = str(uuid.uuid4())   # persist this per user/conversation

r1 = sm.run_with_memory("I'm building a FastAPI app.", session_id=session)
r2 = sm.run_with_memory("What framework am I using?",  session_id=session)

print(r2["output"])  # "You're using FastAPI."

# Async version
r3 = await sm.arun_with_memory("Add a JWT auth endpoint.", session_id=session)

Python Error Types

pythonCopyfrom sentinelmeshai import SentinelMeshAI, AuthError, RateLimitError, SentinelMeshAIError
import time

sm = SentinelMeshAI()

try:
    result = sm.run("What is 2 + 2?")
except AuthError:
    print("Check your SENTINELMESHAI_API_KEY environment variable.")
except RateLimitError:
    time.sleep(2)
    result = sm.run("What is 2 + 2?")   # retry once
except SentinelMeshAIError as e:
    print(f"API error {e.status_code}: {e}")

JavaScript / TypeScript Quickstart

typescriptCopyimport SentinelMeshAI from 'sentinelmeshai';

const sm = new SentinelMeshAI({
  apiKey: process.env.SENTINELMESHAI_API_KEY,
});

// Basic call
const result = await sm.run("Explain the CAP theorem.");
console.log(result.output);
console.log(`Cost: $${result.cost_usd.toFixed(6)} | Latency: ${result.latency_ms}ms`);

// Streaming
const stream = await sm.stream("Write a 100-line Node.js web server.");
for await (const chunk of stream) {
  process.stdout.write(chunk);
}

SentinelMeshAI Class (JavaScript)

run()

typescriptCopy// Express.js route
app.post('/chat', async (req, res) => {
  const result = await sm.run(req.body.prompt, {
    strategy:    'standard',
    temperature: 0.3,
    maxTokens:   2048,
    metadata:    { userId: req.user.id },
  });

  res.json({
    answer:    result.output,
    cost:      result.cost_usd,
    latency:   result.latency_ms,
    model:     result.models_used[0],
  });
});

stream()

typescriptCopy// Next.js App Router streaming response
import { NextRequest } from 'next/server';
import SentinelMeshAI from 'sentinelmeshai';

const sm = new SentinelMeshAI();

export async function POST(request: NextRequest) {
  const { prompt } = await request.json();

  const encoder = new TextEncoder();
  const readable = new ReadableStream({
    async start(controller) {
      for await (const chunk of sm.stream(prompt)) {
        controller.enqueue(encoder.encode(chunk));
      }
      controller.close();
    },
  });

  return new Response(readable, {
    headers: { 'Content-Type': 'text/event-stream' },
  });
}

runWithMemory()

typescriptCopyimport { v4 as uuidv4 } from 'uuid';

const sessionId = uuidv4();   // one per user / conversation

const r1 = await sm.runWithMemory("I work at Acme Corp.", { sessionId });
const r2 = await sm.runWithMemory("Where do I work?",     { sessionId });
console.log(r2.output);  // "You work at Acme Corp."

TypeScript Types

typescriptCopyimport type { RunResult, RunOptions, StreamOptions } from 'sentinelmeshai';

interface RunResult {
  output:        string;
  models_used:   string[];
  strategy_used: string;
  cost_usd:      number;
  latency_ms:    number;
  confidence:    number;
  cached:        boolean;
  trace_id:      string;
}

interface RunOptions {
  strategy?:    string;
  temperature?: number;
  maxTokens?:   number;
  systemPrompt?: string;
  metadata?:    Record<string, unknown>;
}

// Full async iterator type for streaming
type StreamResult = AsyncIterable<string>;

JavaScript Error Types

typescriptCopyimport { AuthError, RateLimitError, SentinelMeshAIError } from 'sentinelmeshai';

try {
  const result = await sm.run("Hello");
} catch (err) {
  if (err instanceof AuthError) {
    console.error("Invalid API key");
  } else if (err instanceof RateLimitError) {
    console.error("Rate limit — wait before retrying");
    await new Promise(r => setTimeout(r, 2000));
  } else if (err instanceof SentinelMeshAIError) {
    console.error(`API error ${err.statusCode}: ${err.message}`);
  } else {
    throw err;  // network error — rethrow
  }
}

Batch Requests

Process multiple prompts concurrently using async gather (Python) or Promise.all (Node.js). SentinelMeshAI handles per-tenant rate limiting server-side — stay within your plan's RPM.

pythonCopyimport asyncio, sentinelmeshai as sm

sm.api_key = os.getenv("SENTINELMESHAI_API_KEY")

async def process_batch(prompts: list[str]):
    # Concurrent — all requests fire simultaneously
    tasks   = [sm.arun(p, strategy="fast_cheap") for p in prompts]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    for prompt, result in zip(prompts, results):
        if isinstance(result, Exception):
            print(f"FAILED: {prompt} → {result}")
        else:
            print(f"OK: {result['output'][:60]}...")

prompts = ["Translate 'hello' to French", "What is 7 * 8?", "Capital of Japan?"]
asyncio.run(process_batch(prompts))

javascriptCopyconst prompts = ["Translate 'hello' to French", "What is 7 * 8?", "Capital of Japan?"];

const results = await Promise.allSettled(
  prompts.map(p => sm.run(p, { strategy: 'fast_cheap' }))
);

results.forEach((r, i) => {
  if (r.status === 'fulfilled') console.log(`OK: ${r.value.output}`);
  else                          console.error(`FAILED: ${prompts[i]} → ${r.reason}`);
});

Retry Strategy

pythonCopyimport asyncio
from sentinelmeshai import SentinelMeshAI, RateLimitError, SentinelMeshAIError

sm = SentinelMeshAI()

async def run_with_backoff(prompt: str, max_attempts: int = 4) -> dict:
    for attempt in range(max_attempts):
        try:
            return await sm.arun(prompt)
        except RateLimitError:
            if attempt == max_attempts - 1: raise
            wait = 2 ** attempt  # 1s, 2s, 4s, 8s
            print(f"Rate limited. Retrying in {wait}s…")
            await asyncio.sleep(wait)
        except SentinelMeshAIError as e:
            if e.status_code >= 500:
                await asyncio.sleep(2 ** attempt)
            else:
                raise   # 4xx — don't retry
    raise RuntimeError("Max retries exceeded")

Testing & Mocking

pythonCopy# pytest + respx (httpx mock library)
import pytest, respx, httpx, sentinelmeshai as sm

@pytest.fixture(autouse=True)
def mock_api(respx_mock):
    respx_mock.post("https://api.sentinelmeshai.com/run").mock(
        return_value=httpx.Response(200, json={
            "output":        "Mocked response",
            "models_used":   ["mock:test"],
            "cost_usd":      0.0,
            "latency_ms":    5,
            "confidence":    1.0,
            "cached":        False,
            "trace_id":      "tr_test",
            "strategy_used": "standard",
        })
    )

def test_my_feature():
    sm.api_key = "sm_test_mock_key"
    result = sm.run("Hello")
    assert result["output"] == "Mocked response"
    assert result["cost_usd"] == 0.0

Framework Integrations

Django middleware

pythonCopy# myapp/utils.py — reusable helper
import os, sentinelmeshai as sm
from django.conf import settings

_client = None

def get_sm_client():
    global _client
    if _client is None:
        _client = sm.SentinelMeshAI(
            api_key  = os.getenv("SENTINELMESHAI_API_KEY"),
            base_url = getattr(settings, "SENTINELMESHAI_URL", "https://api.sentinelmeshai.com"),
        )
    return _client

# Django async view
from django.http import JsonResponse
from asgiref.sync import sync_to_async

async def ai_endpoint(request):
    client = get_sm_client()
    result = await client.arun(request.POST.get("prompt", ""))
    return JsonResponse({"output": result["output"]})

Next.js App Router

typescriptCopy// lib/sm.ts — singleton client
import SentinelMeshAI from 'sentinelmeshai';

export const sm = new SentinelMeshAI({
  apiKey: process.env.SENTINELMESHAI_API_KEY!,
});

// app/api/chat/route.ts
import { sm } from '@/lib/sm';
import { NextRequest, NextResponse } from 'next/server';

export async function POST(req: NextRequest) {
  const { prompt } = await req.json();
  const result = await sm.run(prompt, { strategy: 'standard' });
  return NextResponse.json(result);
}

SDK Changelog

📦 SentinelMeshAI SDK · PyPI · npm · API Docs · Tutorials