Getting started with LangGraph

import express from 'express';
import jwt from 'jsonwebtoken';

const app = express();

const apiKey = process.env.ABLY_API_KEY;
if (!apiKey) {
  throw new Error('ABLY_API_KEY environment variable is required');
}

const [keyName, keySecret] = apiKey.split(':');
if (!keyName || !keySecret) {
  throw new Error('ABLY_API_KEY must be in format "keyName:keySecret"');
}

app.get('/api/auth/token', (req, res) => {
  // In production, authenticate the user and get their ID from your session
  const userId = 'user-123';

  const token = jwt.sign({
    'x-ably-clientId': userId,
    'ably.channel.*': 'user'
  }, keySecret, {
    algorithm: 'HS256',
    keyid: keyName,
    expiresIn: '1h'
  });

  res.type('application/jwt').send(token);
});

app.listen(3001, () => {
  console.log('Auth server running on http://localhost:3001');
});

import * as Ably from 'ably';
import { ChatAnthropic } from '@langchain/anthropic';
import { tool } from '@langchain/core/tools';
import { StateGraph, Annotation, START, END } from '@langchain/langgraph';
import { AIMessage } from '@langchain/core/messages';
import { z } from 'zod';

const apiKey = process.env.ABLY_API_KEY;
if (!apiKey) {
  throw new Error('ABLY_API_KEY environment variable is required');
}

const realtime = new Ably.Realtime({
  key: apiKey,
  clientId: 'ai-agent',
  echoMessages: false,
});

const channel = realtime.channels.get('ai:conversation');

// Define a tool that requires human approval
const sendEmail = tool(
  async ({ to, subject, body }) => {
    console.log(`Sending email to ${to}: ${subject}`);
    return JSON.stringify({ success: true, message: `Email sent to ${to}` });
  },
  {
    name: 'send_email',
    description: 'Send an email to a recipient. Always requires human approval.',
    schema: z.object({
      to: z.string().describe('Recipient email address'),
      subject: z.string().describe('Email subject line'),
      body: z.string().describe('Email body content'),
    }),
  },
);

// Initialize the model with tools
const model = new ChatAnthropic({ model: 'claude-sonnet-4-5' });
const modelWithTools = model.bindTools([sendEmail]);

// Track pending approval requests
const pendingApprovals = new Map<string, (decision: string) => void>();

// Listen for approval responses from users
await channel.subscribe('approval-response', (message: Ably.Message) => {
  const toolCallId = message.extras?.headers?.toolCallId;
  const resolve = pendingApprovals.get(toolCallId);
  if (resolve) {
    pendingApprovals.delete(toolCallId);
    resolve(message.data.decision);
  }
});

// Request human approval for a tool call via the channel
function requestApproval(
  toolCallId: string,
  toolName: string,
  toolInput: Record<string, unknown>,
): Promise<string> {
  return new Promise<string>((resolve) => {
    pendingApprovals.set(toolCallId, resolve);
    channel.publish({
      name: 'approval-request',
      data: { name: toolName, arguments: toolInput },
      extras: { headers: { toolCallId } },
    });
    console.log(`Awaiting approval for ${toolName} (${toolCallId})`);
  });
}

// Stream LangGraph response tokens to Ably
async function streamToAbly(
  messages: any[],
  serial: string,
  useTools: boolean,
) {
  const StateAnnotation = Annotation.Root({
    messages: Annotation<any[]>({
      reducer: (x, y) => x.concat(y),
      default: () => [],
    }),
  });

  const graph = new StateGraph(StateAnnotation)
    .addNode('agent', async (state) => {
      const response = await (useTools ? modelWithTools : model).invoke(state.messages);
      return { messages: [response] };
    })
    .addEdge(START, 'agent')
    .addEdge('agent', END);

  const app = graph.compile();

  const stream = await app.stream(
    { messages },
    { streamMode: 'messages' },
  );

  let toolCallId: string | undefined;
  let toolCallName: string | undefined;
  let toolCallArgsStr = '';

  for await (const [messageChunk, metadata] of stream) {
    const content = messageChunk?.content;

    // Stream text tokens to Ably
    // Content may be a string or an array of content blocks (Anthropic format)
    if (typeof content === 'string' && content) {
      channel.appendMessage({ serial, data: content });
    } else if (Array.isArray(content)) {
      for (const block of content) {
        if (block.type === 'text' && block.text) {
          channel.appendMessage({ serial, data: block.text });
        }
      }
    }

    // Accumulate tool call info from streaming chunks
    for (const chunk of messageChunk?.tool_call_chunks ?? []) {
      if (chunk.id) toolCallId = chunk.id;
      if (chunk.name) toolCallName = chunk.name;
      if (chunk.args) toolCallArgsStr += chunk.args;
    }
  }

  const toolCallDetected = toolCallId && toolCallName
    ? { id: toolCallId, name: toolCallName, args: toolCallArgsStr ? JSON.parse(toolCallArgsStr) : {} }
    : null;

  return { hasToolCall: !!toolCallDetected, toolCallInfo: toolCallDetected };
}

// Handle incoming user prompts
await channel.subscribe('user-input', async (message: Ably.Message) => {
  const { promptId, text } = message.data as { promptId: string; text: string };
  const userId = message.clientId;
  const role = message.extras?.userClaim;

  console.log(`Received prompt from ${userId} (role: ${role}): ${text}`);

  if (role !== 'user') {
    console.log('Ignoring message from non-user');
    return;
  }

  // Create the initial Ably message for streaming
  const response = await channel.publish({
    name: 'agent-response',
    data: '',
    extras: { headers: { promptId } },
  });

  const serial = response.serials[0];
  if (!serial) {
    console.error('No serial returned from publish');
    return;
  }

  // Stream the response from LangGraph
  const messages = [{ role: 'user', content: text }];
  const { hasToolCall, toolCallInfo } = await streamToAbly(messages, serial, true);

  // Handle tool call with human-in-the-loop approval
  if (hasToolCall && toolCallInfo) {
    const decision = await requestApproval(
      toolCallInfo.id,
      toolCallInfo.name,
      toolCallInfo.args,
    );

    let toolResult: string;
    if (decision === 'approved') {
      const result = await sendEmail.invoke(toolCallInfo.args);
      toolResult = result;
    } else {
      toolResult = JSON.stringify({ error: 'The user rejected this action' });
    }

    // Stream follow-up response with the tool result
    channel.appendMessage({ serial, data: '\n\n' });

    const followUpMessages = [
      { role: 'user', content: text },
      new AIMessage({
        content: '',
        tool_calls: [{
          id: toolCallInfo.id,
          name: toolCallInfo.name,
          args: toolCallInfo.args,
        }],
      }),
      { role: 'tool', content: toolResult, tool_call_id: toolCallInfo.id },
    ];

    await streamToAbly(followUpMessages, serial, false);
  }

  // Signal completion
  await channel.publish({
    name: 'agent-response-complete',
    extras: { headers: { promptId } },
  });

  console.log(`Completed response for prompt ${promptId}`);
});

console.log('Agent is listening for prompts...');

import * as Ably from 'ably';
import crypto from 'crypto';
import * as readline from 'readline';

const realtime = new Ably.Realtime({
  authCallback: async (
    _tokenParams: Ably.TokenParams,
    callback: (error: Ably.ErrorInfo | string | null, token: Ably.TokenDetails | Ably.TokenRequest | string | null) => void
  ) => {
    try {
      const response = await fetch('http://localhost:3001/api/auth/token');
      const token = await response.text();
      callback(null, token);
    } catch (error) {
      callback(error instanceof Error ? error.message : String(error), null);
    }
  }
});

realtime.connection.on('connected', () => {
  console.log('Connected to Ably as', realtime.auth.clientId);
});

const channel = realtime.channels.get('ai:conversation');
const pendingPrompts = new Map<string, () => void>();

const rl = readline.createInterface({
  input: process.stdin,
  output: process.stdout,
});

// Subscribe to streamed agent responses
await channel.subscribe('agent-response', (message: Ably.Message) => {
  switch (message.action) {
    case 'message.create':
      break;
    case 'message.append':
      // Write each new token as it arrives
      process.stdout.write(message.data || '');
      break;
    case 'message.update':
      // Full response after reconnection
      console.log(message.data || '');
      break;
  }
});

// Subscribe to approval requests for human-in-the-loop
await channel.subscribe('approval-request', async (message: Ably.Message) => {
  const { name, arguments: args } = message.data;
  const toolCallId = message.extras?.headers?.toolCallId;

  console.log(`\n\nAgent wants to execute: ${name}`);
  console.log(`Arguments: ${JSON.stringify(args, null, 2)}`);

  const answer = await new Promise<string>((resolve) => {
    rl.question('Approve? (yes/no): ', resolve);
  });

  const decision = answer.toLowerCase() === 'yes' ? 'approved' : 'rejected';

  await channel.publish({
    name: 'approval-response',
    data: { decision },
    extras: { headers: { toolCallId } },
  });

  console.log(`Decision sent: ${decision}\n`);
});

// Subscribe to completion signals
await channel.subscribe('agent-response-complete', (message: Ably.Message) => {
  const promptId = message.extras?.headers?.promptId;
  if (!promptId) return;

  console.log('\n');
  const resolve = pendingPrompts.get(promptId);
  if (resolve) {
    pendingPrompts.delete(promptId);
    resolve();
  }
});

async function sendPrompt(text: string): Promise<void> {
  const promptId = crypto.randomUUID();

  const completionPromise = new Promise<void>((resolve) => {
    pendingPrompts.set(promptId, resolve);
  });

  await channel.publish('user-input', {
    promptId,
    text,
  });

  await completionPromise;
}

function askQuestion() {
  rl.question('Enter a prompt (or "quit" to exit): ', async (text) => {
    if (text.toLowerCase() === 'quit') {
      rl.close();
      realtime.close();
      return;
    }

    await sendPrompt(text);
    askQuestion();
  });
}

askQuestion();