Reply to messages with Anthropic API (using threads)
In this section, you'll integrate the Anthropic API with your Zoom Chatbot to handle user messages in threads. When a user sends a message to the bot, your app will generate an AI-powered reply (using models like Claude 3.5) and respond in the same thread.
By the end of this section, you will have:
- A working integration with the Anthropic API
- Threaded AI responses in Chat
- Session-based message history for improved conversational context
Add Anthropic integration
Create a utility function that sends a user's message to Anthropic, receives a completion response, and replies to the chat thread. This function maintains per-user conversation history to support contextual conversations.
File Path
utils/anthropic.js
Code Snippet
import dotenv from "dotenv";
import { sendChatMessage } from "./zoom-api.js";
dotenv.config();
let conversationHistory = new Map();
const decoder = new TextDecoder();
let buf = "";
/**
* Main entry point for interacting with Anthropic API
* @param {Object} payload - Zoom webhook payload containing toJid / message
* @param {Object} options - { stream?: boolean, onStreamChunk?: (chunk, soFar) => void }
*/
export async function callAnthropicAPI(payload, options = {}) {
const userJid = payload?.toJid;
if (!userJid) {
console.error("Error: payload.toJid is missing.");
return;
}
try {
const { stream = true, onStreamChunk = null } = options;
const headers = buildHeaders();
// Build conversation history
const history = conversationHistory.get(userJid) || [];
const userMessage = {
role: "user",
content: payload.cmd || payload.message || "Hello",
};
history.push(userMessage);
const requestData = buildRequestBody({
model: process.env.ANTHROPIC_MODEL || "claude-3-5-sonnet-latest",
messages: history,
stream,
});
console.log(
`Sending message to Anthropic (model=${requestData.model}, stream=${stream}) for user: ${userJid}`,
);
const completion = stream
? await handleStreamingResponse(
requestData,
headers,
userJid,
payload,
onStreamChunk,
)
: await handleNonStreamingResponse(
requestData,
headers,
userJid,
payload,
);
return completion;
} catch (error) {
console.error("Error in callAnthropicAPI:", error?.message || error);
try {
await sendChatMessage(
payload?.toJid,
"Sorry, I hit an AI model error. I will be back shortly. (Check model access/config.)",
payload?.reply_to || null,
);
} catch (sendError) {
console.error(
"Failed to send error message to user:",
sendError?.message || sendError,
);
}
throw error;
}
}
function buildHeaders() {
const apiKey = process.env.ANTHROPIC_API_KEY;
if (!apiKey)
throw new Error("Missing ANTHROPIC_API_KEY in environment variables.");
if (!apiKey.startsWith("sk-ant-"))
throw new Error(
'Invalid ANTHROPIC_API_KEY format. Should start with "sk-ant-".',
);
return {
"Content-Type": "application/json",
"anthropic-version": "2023-06-01",
"x-api-key": apiKey,
};
}
function buildRequestBody({ model, messages, stream }) {
return {
model,
messages,
system: "You are a helpful AI assistant integrated with Chat. Provide concise, helpful responses to user questions and requests.",
temperature: 0.7,
max_tokens: 1000,
};
}
/**
* Non-streaming path using native fetch.
*/
async function handleNonStreamingResponse(
requestData,
headers,
userJid,
payload,
) {
const ANTHROPIC_URL =
"[https://api.anthropic.com/v1/messages](https://api.anthropic.com/v1/messages)";
const response = await fetch(ANTHROPIC_URL, {
method: "POST",
headers,
body: JSON.stringify({ ...requestData, stream: false }),
});
if (!response.ok) {
let errorData = {};
try {
errorData = await response.json();
} catch {
errorData = { message: await response.text() };
}
const error = new Error(
`HTTP ${response.status}: ${response.statusText}`,
);
error.status = response.status;
error.response = { data: errorData };
throw error;
}
const data = await response.json();
if (!data?.content || !Array.isArray(data.content)) {
throw new Error(
`Unexpected response from Anthropic API: ${JSON.stringify(data)}`,
);
}
const completion = data.content
.filter((b) => b.type === "text")
.map((b) => b.text)
.join("\n")
.trim();
if (!completion) throw new Error("Empty response from Anthropic API");
// Append assistant message & trim history
const history = conversationHistory.get(userJid) || [];
history.push({ role: "assistant", content: completion });
conversationHistory.set(
userJid,
history.length > 20 ? history.slice(-20) : history,
);
await sendChatMessage(userJid, completion, payload.reply_to || null);
console.log("Successfully sent response to Chat");
return completion;
}
/**
* Streaming path using native fetch + Web Streams.
* Parses Server-Sent Events (SSE): frames separated by blank line, with "event:" and "data:" lines.
*/
async function handleStreamingResponse(
requestData,
headers,
userJid,
payload,
onStreamChunk,
) {
const ANTHROPIC_URL =
"[https://api.anthropic.com/v1/messages](https://api.anthropic.com/v1/messages)";
const sseHeaders = { ...headers, Accept: "text/event-stream" };
const response = await fetch(ANTHROPIC_URL, {
method: "POST",
headers: sseHeaders,
body: JSON.stringify({ ...requestData, stream: true }),
});
if (!response.ok) {
let errorData = {};
try {
errorData = await response.json();
} catch {
errorData = { message: await response.text() };
}
const error = new Error(
`HTTP ${response.status}: ${response.statusText}`,
);
error.status = response.status;
error.response = { data: errorData };
throw error;
}
const reader = response.body.getReader();
let completion = "";
try {
for (;;) {
const { done, value } = await reader.read();
if (done) break;
buf += decoder.decode(value, { stream: true });
// SSE frames are separated by a blank line (\n\n)
let sep;
while ((sep = buf.indexOf("\n\n")) !== -1) {
const frame = buf.slice(0, sep);
buf = buf.slice(sep + 2);
let eventType = "";
let dataLine = "";
for (const line of frame.split("\n")) {
if (line.startsWith("event:"))
eventType = line.slice(6).trim();
else if (line.startsWith("data:"))
dataLine += line.slice(5).trim();
}
if (!dataLine) continue;
if (dataLine === "[DONE]") continue;
try {
const parsed = JSON.parse(dataLine);
if (
parsed.type === "content_block_delta" &&
parsed.delta?.type === "text_delta"
) {
const chunk = parsed.delta.text || "";
if (chunk) {
completion += chunk;
onStreamChunk?.(chunk, completion);
}
}
if (
parsed.type === "message_stop" ||
eventType === "message_stop"
) {
break;
}
} catch {
// Ignore malformed pieces
}
}
}
} finally {
reader.releaseLock();
}
if (!completion.trim()) {
buf = "";
throw new Error("Empty response from Anthropic streaming API");
}
// Append assistant message & trim history
const history = conversationHistory.get(userJid) || [];
history.push({ role: "assistant", content: completion });
conversationHistory.set(
userJid,
history.length > 20 ? history.slice(-20) : history,
);
// Send the full completion to Zoom
await sendChatMessage(userJid, completion, payload.reply_to || null);
console.log("Successfully sent streaming response to Chat");
buf = "";
return completion;
}
export async function callAnthropicAPIStreaming(payload, onStreamChunk) {
return callAnthropicAPI(payload, {
stream: true,
onStreamChunk,
});
}
export function getConversationHistory(userJid) {
return conversationHistory.get(userJid) || [];
}
export function clearConversationHistory(userJid) {
conversationHistory.delete(userJid);
console.log(`Cleared conversation history for user: ${userJid}`);
}
Update webhook to trigger Anthropic replies
Modify the bot_notification case in your webhook handler to call the Anthropic API whenever a user sends a message to the chatbot.
This implementation enables your Zoom chatbot to respond in threads using AI-generated answers, creating a conversational experience powered by Anthropic's large language models.
File Path
routes/zoom-webhookHandler.js
Code Snippet
import { callAnthropicAPI } from '../utils/anthropic.js';
// ... inside your webhook router switch statement ...
case 'bot_notification':
console.log('Processing bot notification from Zoom Team Chat');
await callAnthropicAPI(payload, true);
break;