From zero to production in minutes.

from airmy import Airmy

client = Airmy(api_key="sk-airmy-...")

run = client.runs.create(
    agent_id="ag_01HXYZ",
    input={"query": "Hello, world!"}
)

print(run.output)
# {"result": "Hello! How can I help you today?"}

import asyncio
from airmy import AsyncAirmy

async def process_docs(docs: list[str]):
    client = AsyncAirmy(api_key="sk-airmy-...")

    tasks = [
        client.runs.create(
            agent_id="ag_doc_summariser",
            input={"text": doc}
        )
        for doc in docs
    ]

    results = await asyncio.gather(*tasks)
    return [r.output["summary"] for r in results]

summaries = asyncio.run(process_docs(documents))

import Airmy from '@airmy/sdk';

const client = new Airmy({ apiKey: process.env.AIRMY_API_KEY });

async function pipeline(input: string) {
  // Step 1: Classify
  const intent = await client.runs.create({
    agentId: 'ag_classifier',
    input: { text: input }
  });

  // Step 2: Analyse
  const analysis = await client.runs.create({
    agentId: 'ag_analyst',
    input: { intent: intent.output, raw: input }
  });

  // Step 3: Report
  return client.runs.create({
    agentId: 'ag_reporter',
    input: { analysis: analysis.output }
  });
}

from airmy import Airmy

client = Airmy(api_key="sk-airmy-...")

with client.runs.stream(
    agent_id="ag_01HXYZ",
    input={"query": "Summarise Q1 financials"}
) as stream:
    for chunk in stream.text_stream:
        print(chunk, end="", flush=True)

    final = stream.get_final_run()
    print(f"\n\nTokens used: {final.tokens_used}")

from fastapi import FastAPI, Request, HTTPException
import hmac, hashlib, os

app = FastAPI()
SECRET = os.environ["AIRMY_WEBHOOK_SECRET"]

@app.post("/hooks/airmy")
async def handle_webhook(request: Request):
    body = await request.body()
    sig = request.headers.get("X-AIRMY-Signature", "")
    expected = hmac.new(SECRET.encode(), body, hashlib.sha256).hexdigest()

    if not hmac.compare_digest(expected, sig):
        raise HTTPException(status_code=401)

    event = await request.json()

    if event["type"] == "run.completed":
        print("Run completed:", event["data"]["id"])

    return {"ok": True}

import Airmy, { AirmyError, RateLimitError } from '@airmy/sdk';

const client = new Airmy({ apiKey: process.env.AIRMY_API_KEY });

async function runWithFallback(input: object) {
  try {
    return await client.runs.create({
      agentId: 'ag_primary',
      input
    });
  } catch (err) {
    if (err instanceof RateLimitError) {
      await new Promise(r => setTimeout(r, err.retryAfter * 1000));
      return runWithFallback(input); // retry once
    }
    if (err instanceof AirmyError && err.status >= 500) {
      // fallback to secondary agent
      return client.runs.create({ agentId: 'ag_fallback', input });
    }
    throw err;
  }
}

# 1. Exchange client credentials for session token
curl -X POST https://api.airmy.dev/v2/auth/token \
  -H "Content-Type: application/json" \
  -d '{
    "grant_type": "client_credentials",
    "client_id": "'"$AIRMY_CLIENT_ID"'",
    "client_secret": "'"$AIRMY_CLIENT_SECRET"'",
    "scope": "runs:write agents:read"
  }'

# Response:
# { "access_token": "ey...", "expires_in": 3600 }

# 2. Use the token
curl -X POST https://api.airmy.dev/v2/runs \
  -H "Authorization: Bearer $ACCESS_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"agent_id": "ag_01HXYZ", "input": {"query": "..."}}'

from airmy import Airmy

client = Airmy(api_key="sk-airmy-...")

# Create an alert rule
alert = client.alerts.create(
    agent_id="ag_01HXYZ",
    condition="error_rate_5m > 0.05",  # > 5% error rate
    channels=["slack:#ai-ops", "pagerduty:oncall"],
    severity="high",
    name="High Error Rate — data-engineer-v2"
)

print(f"Alert created: {alert.id}")

# Query current metrics
metrics = client.metrics.query(
    agent_id="ag_01HXYZ",
    metrics=["error_rate", "p99_latency", "calls_per_min"],
    window="5m"
)

for m in metrics:
    print(f"{m.name}: {m.value}")