Recordings & transcripts

curl "https://agents.assemblyai.com/v1/sessions?limit=50&agent_id=$AGENT_ID" \
  -H "Authorization: $ASSEMBLYAI_API_KEY"

# pip install requests
import os
import requests

resp = requests.get(
    "https://agents.assemblyai.com/v1/sessions",
    headers={"Authorization": os.environ["ASSEMBLYAI_API_KEY"]},
    params={"limit": 50, "agent_id": os.environ["AGENT_ID"]},
)
resp.raise_for_status()
print(resp.json()["sessions"])

// Node 18+ has fetch built in
const url = new URL("https://agents.assemblyai.com/v1/sessions");
url.searchParams.set("limit", "50");
url.searchParams.set("agent_id", process.env.AGENT_ID);

const res = await fetch(url, {
  headers: { Authorization: process.env.ASSEMBLYAI_API_KEY },
});
const data = await res.json();
console.log(data.sessions);

import os
import requests

BASE = "https://agents.assemblyai.com"
HEADERS = {"Authorization": os.environ["ASSEMBLYAI_API_KEY"]}


def all_sessions(**filters):
    cursor = None
    while True:
        resp = requests.get(
            f"{BASE}/v1/sessions",
            headers=HEADERS,
            params={"limit": 200, "cursor": cursor, **filters},
        )
        resp.raise_for_status()
        page = resp.json()
        yield from page["sessions"]
        cursor = page["response_metadata"]["next_cursor"]
        if not page["has_more"] or not cursor:
            break


for session in all_sessions(agent_id=os.environ["AGENT_ID"]):
    print(session["id"], session["status"], session["duration_seconds"])

const BASE = "https://agents.assemblyai.com";
const HEADERS = { Authorization: process.env.ASSEMBLYAI_API_KEY };

async function* allSessions(filters = {}) {
  let cursor = "";
  do {
    const url = new URL(`${BASE}/v1/sessions`);
    url.searchParams.set("limit", "200");
    if (cursor) url.searchParams.set("cursor", cursor);
    for (const [k, v] of Object.entries(filters)) url.searchParams.set(k, v);

    const res = await fetch(url, { headers: HEADERS });
    const page = await res.json();
    yield* page.sessions;
    cursor = page.response_metadata.next_cursor;
    if (!page.has_more || !cursor) break;
  } while (true);
}

for await (const s of allSessions({ agent_id: process.env.AGENT_ID })) {
  console.log(s.id, s.status, s.duration_seconds);
}

curl "https://agents.assemblyai.com/v1/sessions/$SESSION_ID" \
  -H "Authorization: $ASSEMBLYAI_API_KEY"

import os
import requests

resp = requests.get(
    f"https://agents.assemblyai.com/v1/sessions/{os.environ['SESSION_ID']}",
    headers={"Authorization": os.environ["ASSEMBLYAI_API_KEY"]},
)
resp.raise_for_status()
session = resp.json()
print(session["status"], session["artifacts"])

const res = await fetch(
  `https://agents.assemblyai.com/v1/sessions/${process.env.SESSION_ID}`,
  { headers: { Authorization: process.env.ASSEMBLYAI_API_KEY } },
);
const session = await res.json();
console.log(session.status, session.artifacts);

import os
import requests

BASE = "https://agents.assemblyai.com"
HEADERS = {"Authorization": os.environ["ASSEMBLYAI_API_KEY"]}


def artifact_url(session, kind):
    return next((a["url"] for a in session["artifacts"] if a["type"] == kind), None)


def load_transcript(session_id):
    session = requests.get(
        f"{BASE}/v1/sessions/{session_id}", headers=HEADERS
    ).json()

    url = artifact_url(session, "timeline")
    if url is None:
        return []  # session still active, or no recording

    timeline = requests.get(url).json()  # pre-signed URL: no auth header

    messages = []
    for turn in timeline.get("turns", []):  # `turns` is omitted when empty
        if turn.get("user_transcript"):
            messages.append({"role": "user", "text": turn["user_transcript"]})
        for call in turn.get("tool_calls", []):
            messages.append({
                "role": "tool",
                "name": call["name"],
                "arguments": call.get("arguments"),
                "result": call.get("result"),
                "error": call["is_error"] or call.get("timed_out", False),
            })
        if turn.get("agent_text"):
            messages.append({"role": "agent", "text": turn["agent_text"]})
    return messages


for m in load_transcript(os.environ["SESSION_ID"]):
    if m["role"] == "tool":
        flag = " (error)" if m["error"] else ""
        print(f"  ↳ {m['name']}({m['arguments']}) → {m['result']}{flag}")
    else:
        print(f"{m['role']}: {m['text']}")

const BASE = "https://agents.assemblyai.com";
const HEADERS = { Authorization: process.env.ASSEMBLYAI_API_KEY };

const artifactUrl = (session, kind) =>
  session.artifacts.find((a) => a.type === kind)?.url ?? null;

async function loadTranscript(sessionId) {
  const session = await (
    await fetch(`${BASE}/v1/sessions/${sessionId}`, { headers: HEADERS })
  ).json();

  const url = artifactUrl(session, "timeline");
  if (!url) return []; // session still active, or no recording

  const timeline = await (await fetch(url)).json(); // pre-signed: no auth header

  const messages = [];
  for (const turn of timeline.turns ?? []) {  // `turns` is omitted when empty
    if (turn.user_transcript)
      messages.push({ role: "user", text: turn.user_transcript });
    for (const call of turn.tool_calls ?? []) {
      messages.push({
        role: "tool",
        name: call.name,
        arguments: call.arguments,
        result: call.result,
        error: call.is_error || call.timed_out,
      });
    }
    if (turn.agent_text)
      messages.push({ role: "agent", text: turn.agent_text });
  }
  return messages;
}

for (const m of await loadTranscript(process.env.SESSION_ID)) {
  if (m.role === "tool") {
    console.log(
      `  ↳ ${m.name}(${JSON.stringify(m.arguments)}) → ${m.result}` +
        (m.error ? " (error)" : ""),
    );
  } else {
    console.log(`${m.role}: ${m.text}`);
  }
}

import os
import requests

BASE = "https://agents.assemblyai.com"
HEADERS = {"Authorization": os.environ["ASSEMBLYAI_API_KEY"]}

session = requests.get(
    f"{BASE}/v1/sessions/{os.environ['SESSION_ID']}", headers=HEADERS
).json()

audio = next(a for a in session["artifacts"] if a["type"] == "audio")
meta = next(a for a in session["artifacts"] if a["type"] == "metadata")

with requests.get(audio["url"], stream=True) as r:  # pre-signed: no auth header
    r.raise_for_status()
    with open("recording.ogg", "wb") as f:
        for chunk in r.iter_content(chunk_size=8192):
            f.write(chunk)

print(requests.get(meta["url"]).json())

import { writeFile } from "node:fs/promises";

const BASE = "https://agents.assemblyai.com";
const HEADERS = { Authorization: process.env.ASSEMBLYAI_API_KEY };

const session = await (
  await fetch(`${BASE}/v1/sessions/${process.env.SESSION_ID}`, {
    headers: HEADERS,
  })
).json();

const audio = session.artifacts.find((a) => a.type === "audio");
const meta = session.artifacts.find((a) => a.type === "metadata");

const buf = Buffer.from(await (await fetch(audio.url)).arrayBuffer());
await writeFile("recording.ogg", buf);

console.log(await (await fetch(meta.url)).json());

import express from "express";

const app = express();

app.get("/api/sessions/:id/audio", async (req, res) => {
  const r = await fetch(
    `https://agents.assemblyai.com/v1/sessions/${req.params.id}`,
    { headers: { Authorization: process.env.ASSEMBLYAI_API_KEY } },
  );
  if (!r.ok) return res.status(r.status).end();

  const session = await r.json();
  const audio = session.artifacts.find((a) => a.type === "audio");
  // The URL is short-lived, so the client requests it right before playback.
  res.json({ url: audio?.url ?? null });
});

app.listen(3000);

import { useEffect, useState } from "react";

export function SessionRecording({ sessionId }: { sessionId: string }) {
  const [url, setUrl] = useState<string | null>(null);

  useEffect(() => {
    fetch(`/api/sessions/${sessionId}/audio`)
      .then((r) => r.json())
      .then((d) => setUrl(d.url));
  }, [sessionId]);

  if (!url) return <p>Loading recording…</p>;

  return (
    <audio controls preload="metadata" src={url}>
      Your browser does not support the audio element.
    </audio>
  );
}