SentinelMeshAI Tutorials

Sign Up

1. Go to app.sentinelmeshai.com
2. Click Sign Up — use email or continue with GitHub / Google
3. Verify your email (check spam if needed)
4. You land on the Dashboard — you're in

Get Your API Key

1. In the dashboard, click API Keys in the left sidebar
2. Click + New Key
3. Name it "Tutorial Key" and select Live
4. Click Generate Key
5. Copy the full key (starts with sm_live_) — it is shown only once
6. Paste it somewhere safe (your password manager or a .env file)

Store Your Key Securely

bashCopy# Mac / Linux — add to ~/.zshrc or ~/.bashrc
export SENTINELMESHAI_API_KEY="sm_live_..."

# Windows PowerShell
$env:SENTINELMESHAI_API_KEY="sm_live_..."

# .env file (for projects)
echo 'SENTINELMESHAI_API_KEY=sm_live_...' > .env

Test the Key

bashCopycurl -X GET https://api.sentinelmeshai.com/health \
  -H "Authorization: Bearer $SENTINELMESHAI_API_KEY"

{"status": "healthy", "version": "1.0.0", "tenant": "your-name"}

Install

bashCopypip install sentinelmeshai

bashCopynpm install sentinelmeshai

Hello, SentinelMeshAI

pythonCopyimport os, sentinelmeshai as sm

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

result = sm.run("What are three interesting facts about octopuses?")

print(result["output"])
print()
print(f"Model used : {result['models_used']}")
print(f"Cost       : ${result['cost_usd']:.6f}")
print(f"Latency    : {result['latency_ms']}ms")
print(f"Cached     : {result['cached']}")

javascriptCopyimport SentinelMeshAI from 'sentinelmeshai';

const sm = new SentinelMeshAI();

const result = await sm.run("What are three interesting facts about octopuses?");
console.log(result.output);
console.log(`Model: ${result.models_used[0]}`);
console.log(`Cost: $${result.cost_usd.toFixed(6)}`);

bashCopycurl -X POST https://api.sentinelmeshai.com/run \
  -H "Authorization: Bearer $SENTINELMESHAI_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"prompt": "What are three interesting facts about octopuses?"}'

1. Octopuses have three hearts — two pump blood to the gills, one to the body.
2. Their blood is blue, due to hemocyanin (copper-based), not hemoglobin.
3. Each arm has its own mini "brain" — 2/3 of their neurons live in their arms.

Model used : ['anthropic:claude-3-5-sonnet-20241022']
Cost       : $0.000724
Latency    : 387ms
Cached     : False

Reading the Response Object

Add an AI Route to Django

pythonCopy# myapp/views.py
import os
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST
import sentinelmeshai as sm, json

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

@csrf_exempt
@require_POST
async def ai_answer(request):
    body = json.loads(request.body)
    prompt = body.get("prompt", "").strip()
    if not prompt:
        return JsonResponse({"error": "prompt required"}, status=400)

    result = await sm.arun(
        prompt,
        strategy   = "standard",
        max_tokens = 1024,
        metadata   = {"user_id": str(request.user.id) if request.user.is_authenticated else "anon"},
    )
    return JsonResponse({
        "answer":   result["output"],
        "cost_usd": result["cost_usd"],
        "latency":  result["latency_ms"],
    })

pythonCopy# myapp/urls.py
from django.urls import path
from . import views

urlpatterns = [
    path("api/ai/answer/", views.ai_answer, name="ai_answer"),
]

Add an AI Route to Express

javascriptCopyimport express from 'express';
import SentinelMeshAI from 'sentinelmeshai';

const app = express();
app.use(express.json());

const sm = new SentinelMeshAI();

app.post('/api/ai/answer', async (req, res) => {
  const { prompt } = req.body;
  if (!prompt?.trim()) return res.status(400).json({ error: 'prompt required' });

  try {
    const result = await sm.run(prompt, {
      strategy: 'standard',
      metadata: { userId: req.user?.id },
    });
    res.json({ answer: result.output, cost: result.cost_usd });
  } catch (err) {
    console.error(err);
    res.status(500).json({ error: 'AI service error' });
  }
});

app.listen(3000);

Streaming with FastAPI

pythonCopyfrom fastapi import FastAPI
from fastapi.responses import StreamingResponse
import sentinelmeshai as sm, os

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

@app.post("/stream")
async def stream_endpoint(body: dict):
    prompt = body.get("prompt", "")

    async def token_generator():
        async for chunk in sm.astream(prompt):
            yield f"data: {chunk}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(token_generator(), media_type="text/event-stream")

javascriptCopy// React frontend — consume the stream
async function AskAI({ prompt }) {
  const [answer, setAnswer] = useState('');

  const fetchStream = async () => {
    const resp = await fetch('/stream', {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({ prompt }),
    });
    const reader = resp.body.getReader();
    const decoder = new TextDecoder();

    while (true) {
      const { done, value } = await reader.read();
      if (done) break;
      const text = decoder.decode(value);
      // parse SSE: "data: chunk\n\n"
      for (const line of text.split('\n')) {
        if (line.startsWith('data: ') && !line.includes('[DONE]')) {
          setAnswer(prev => prev + line.slice(6));
        }
      }
    }
  };

  return <div><button onClick={fetchStream}>Ask</button><p>{answer}</p></div>;
}

Persistent Conversation Session

pythonCopyimport uuid, sentinelmeshai as sm, os

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

# One session_id per user/conversation — store in DB or session cookie
def create_session() -> str:
    return f"chat_{uuid.uuid4().hex[:12]}"

async def chat(session_id: str, user_message: str) -> str:
    result = await sm.arun_with_memory(
        user_message,
        session_id  = session_id,
        strategy    = "standard",
        system_prompt = "You are a helpful assistant. Be concise.",
    )
    return result["output"]

# Django view example
async def chatbot_view(request):
    session_id = request.session.get("chat_session") or create_session()
    request.session["chat_session"] = session_id

    user_msg = request.POST.get("message", "")
    response = await chat(session_id, user_msg)

    return JsonResponse({"response": response, "session_id": session_id})

Get notified when approaching limits

pythonCopyimport requests, os

# Register webhook via admin endpoint
requests.post(
    "https://api.sentinelmeshai.com/admin/webhook",
    headers={"X-Admin-Key": os.getenv("SENTINELMESHAI_ADMIN_KEY")},
    json={
        "url":    "https://yourapp.com/webhooks/ai-alerts",
        "events": ["usage.threshold", "rate_limit.hit", "model.error"],
        "secret": "your_webhook_signing_secret",
    }
)

# Webhook receiver (Django view)
import hmac, hashlib, json
from django.http import HttpResponse
from django.views.decorators.csrf import csrf_exempt

@csrf_exempt
def webhook_receiver(request):
    sig = request.headers.get("X-Webhook-Signature", "")
    expected = hmac.new(
        b"your_webhook_signing_secret",
        request.body,
        hashlib.sha256,
    ).hexdigest()

    if not hmac.compare_digest(sig, expected):
        return HttpResponse(status=401)

    event = json.loads(request.body)
    if event["event"] == "usage.threshold":
        # e.g. notify the user, pause non-critical jobs
        print(f"Usage alert: {event['data']['used_pct']}% of monthly credits used")

    return HttpResponse(status=200)

Creating a Multi-Tool Research Agent

pythonCopyimport aiohttp
from core.tools.base_tool    import BaseTool
from core.tools.tool_registry import ToolRegistry
from core.agents.tool_agent   import ToolAgent
from app import router  # global router from app.py


class WebSearchTool(BaseTool):
    name        = "web_search"
    description = "Search the web and return the top 5 results for a query"
    async def execute(self, query: str) -> str:
        # Replace with your real search API (Brave, SerpAPI, etc.)
        async with aiohttp.ClientSession() as s:
            resp = await s.get("https://api.example.com/search",
                                params={"q": query, "count": 5})
            data = await resp.json()
        return "\n".join(f"- {r['title']}: {r['url']}" for r in data["results"])


class CalculateTool(BaseTool):
    name        = "calculate"
    description = "Evaluate a math expression, e.g. '15 * 0.089' or 'sqrt(256)'"
    async def execute(self, expression: str) -> str:
        # Walk the AST instead of calling eval() -- a restricted-builtins
        # eval() *looks* sandboxed but is a well-known escape (any object
        # reachable from the expression can chain through __class__ /
        # __subclasses__ back to os.system). This is the same AST-walking
        # approach SentinelMesh's own built-in `calculate` tool uses.
        import ast, math, operator
        if len(expression) > 500:
            return "Error: expression exceeds 500-character limit."
        BIN_OPS   = {ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul,
                     ast.Div: operator.truediv, ast.Pow: operator.pow, ast.Mod: operator.mod}
        UNARY_OPS = {ast.USub: operator.neg, ast.UAdd: operator.pos}
        FUNCS = {n: getattr(math, n) for n in
                 ("sqrt", "sin", "cos", "tan", "log", "log10", "exp", "floor", "ceil")}

        def _eval(node):
            if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
                return node.value
            if isinstance(node, ast.BinOp) and type(node.op) in BIN_OPS:
                return BIN_OPS[type(node.op)](_eval(node.left), _eval(node.right))
            if isinstance(node, ast.UnaryOp) and type(node.op) in UNARY_OPS:
                return UNARY_OPS[type(node.op)](_eval(node.operand))
            if (isinstance(node, ast.Call) and isinstance(node.func, ast.Name)
                    and node.func.id in FUNCS and len(node.args) == 1 and not node.keywords):
                return FUNCS[node.func.id](_eval(node.args[0]))
            raise ValueError(f"Unsupported expression: {ast.dump(node)}")

        try:
            return str(_eval(ast.parse(expression, mode="eval").body))
        except Exception as e:
            return f"Error: {e}"


class SummariseTool(BaseTool):
    name        = "summarise_url"
    description = "Fetch a URL and return a summary of its content"
    async def execute(self, url: str) -> str:
        async with aiohttp.ClientSession() as s:
            resp = await s.get(url, headers={"User-Agent": "SentinelMeshAIBot/1.0"})
            html = await resp.text()
        # Strip HTML tags (simplified)
        import re
        text = re.sub(r"<[^>]+>", " ", html)[:3000]
        return text.strip()


registry = ToolRegistry()
registry.register(WebSearchTool())
registry.register(CalculateTool())
registry.register(SummariseTool())

agent = ToolAgent("research_agent", router=router, tool_registry=registry)

Executing a Multi-Step Research Task

pythonCopyresult = await agent.run(
    "Research the current price of Nvidia stock, calculate what $10,000 "
    "invested 5 years ago would be worth today if it grew at the same rate, "
    "and summarise the latest news about Nvidia's AI chip business."
)

print(result.output)
print(f"\nTools called : {result.tool_calls_made}")  # ['web_search', 'calculate', 'summarise_url']
print(f"Total cost   : ${result.cost_usd:.4f}")
print(f"Rounds       : {result.rounds}")              # ≤ 5 (loop protection built in)

Round 1 → [tool_call] web_search("Nvidia stock price 2024")
Round 1 → [tool_result] "NVDA: $875.35 (as of...)"
Round 2 → [tool_call] calculate("875.35 / 135.58 * 10000")
Round 2 → [tool_result] "64568.08"
Round 3 → [tool_call] summarise_url("https://reuters.com/technology/nvidia...")
Round 3 → [tool_result] "Nvidia announced H200 GPU availability..."
Round 4 → [final answer] "Based on Nvidia's current price of $875..."

Auto-Review Pull Requests with AI

pythonCopyfrom django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
import hmac, hashlib, json, os
import sentinelmeshai as sm, httpx

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

GITHUB_SECRET = os.getenv("GITHUB_WEBHOOK_SECRET", "").encode()
GITHUB_TOKEN  = os.getenv("GITHUB_TOKEN", "")

@csrf_exempt
async def github_webhook(request):
    # 1. Verify the GitHub signature
    sig = request.headers.get("X-Hub-Signature-256", "")
    expected = "sha256=" + hmac.new(GITHUB_SECRET, request.body, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(sig, expected):
        return JsonResponse({"error": "invalid signature"}, status=401)

    event = json.loads(request.body)
    if event.get("action") not in ("opened", "synchronize"):
        return JsonResponse({"ok": True, "skipped": True})

    # 2. Fetch the diff
    diff_url  = event["pull_request"]["diff_url"]
    pr_number = event["number"]
    repo      = event["repository"]["full_name"]

    async with httpx.AsyncClient() as client:
        diff_resp = await client.get(diff_url,
                                      headers={"Authorization": f"Bearer {GITHUB_TOKEN}"})
        diff = diff_resp.text[:8000]  # truncate for token budget

    # 3. AI review
    review = await sm.arun(
        f"""Review this pull request diff and provide:
1. A summary of what changes were made
2. Any potential bugs or security issues
3. Code quality suggestions
4. An overall verdict (APPROVE / REQUEST_CHANGES)

Diff:
{diff}""",
        strategy    = "code_specialist",
        temperature = 0.1,
    )

    # 4. Post the review back to GitHub
    async with httpx.AsyncClient() as client:
        await client.post(
            f"https://api.github.com/repos/{repo}/issues/{pr_number}/comments",
            headers={"Authorization": f"Bearer {GITHUB_TOKEN}"},
            json={"body": f"🤖 **SentinelMeshAI AI Review**\n\n{review['output']}"},
        )

    return JsonResponse({"ok": True, "cost_usd": review["cost_usd"]})

Process AI Tasks in the Background

pythonCopy# tasks.py
import os, asyncio
from celery import shared_task
import sentinelmeshai as sm

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

@shared_task(bind=True, max_retries=3, default_retry_delay=5)
def classify_support_ticket(self, ticket_id: int, ticket_text: str):
    """Classify a support ticket's urgency and route to the right team."""
    try:
        result = asyncio.run(sm.arun(
            f"""Classify this support ticket:

"{ticket_text}"

Return JSON only:
{{
  "urgency": "critical|high|medium|low",
  "category": "billing|technical|account|feature",
  "sentiment": "angry|frustrated|neutral|positive",
  "one_line_summary": "string"
}}""",
            strategy    = "fast_reasoning",
            temperature = 0.0,
        ))
        import json
        classification = json.loads(result["output"])

        # Update the ticket in DB
        from support.models import Ticket
        ticket = Ticket.objects.get(id=ticket_id)
        ticket.urgency    = classification["urgency"]
        ticket.category   = classification["category"]
        ticket.ai_summary = classification["one_line_summary"]
        ticket.save()

        # Route critical tickets immediately
        if classification["urgency"] == "critical":
            notify_on_call_team.delay(ticket_id)

    except Exception as exc:
        self.retry(exc=exc)


# Fire the task when a ticket is created
from django.db.models.signals import post_save
from support.models import Ticket

@receiver(post_save, sender=Ticket)
def on_ticket_created(sender, instance, created, **kwargs):
    if created:
        classify_support_ticket.delay(instance.id, instance.body)