Getting started with OpenAI

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 OpenAI from 'openai';

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

const openai = new OpenAI();

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 tools: OpenAI.Responses.Tool[] = [
  {
    type: 'function',
    name: 'send_email',
    description: 'Send an email to a recipient. Always requires human approval.',
    parameters: {
      type: 'object',
      properties: {
        to: { type: 'string', description: 'Recipient email address' },
        subject: { type: 'string', description: 'Email subject line' },
        body: { type: 'string', description: 'Email body content' },
      },
      required: ['to', 'subject', 'body'],
    },
  },
];

// 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})`);
  });
}

// Execute a tool after approval
function executeTool(name: string, input: Record<string, unknown>) {
  if (name === 'send_email') {
    console.log(`Sending email to ${input.to}: ${input.subject}`);
    return { success: true, message: `Email sent to ${input.to}` };
  }
  return { error: `Unknown tool: ${name}` };
}

// Stream OpenAI response tokens to Ably, returning tool call info if any
async function streamToAbly(
  input: OpenAI.Responses.ResponseInput,
  serial: string,
) {
  const stream = await openai.responses.create({
    model: 'gpt-4o',
    input,
    tools,
    stream: true,
  });

  let messageItemId: string | null = null;
  let functionCallItem: { id: string; callId: string; name: string } | null = null;
  let functionArgs = '';
  let hasToolCall = false;

  for await (const event of stream) {
    switch (event.type) {
      case 'response.output_item.added':
        if (event.item.type === 'message') {
          messageItemId = event.item.id;
        } else if (event.item.type === 'function_call') {
          functionCallItem = {
            id: event.item.id,
            callId: event.item.call_id,
            name: event.item.name,
          };
          functionArgs = '';
          hasToolCall = true;
        }
        break;

      case 'response.output_text.delta':
        if (event.item_id === messageItemId) {
          channel.appendMessage({ serial, data: event.delta });
        }
        break;

      case 'response.function_call_arguments.delta':
        functionArgs += event.delta;
        break;

      case 'response.completed':
        break;
    }
  }

  return {
    hasToolCall,
    functionCallItem,
    functionArgs,
  };
}

// 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 OpenAI
  const input: OpenAI.Responses.ResponseInput = [
    { role: 'user', content: text },
  ];

  const { hasToolCall, functionCallItem, functionArgs } = await streamToAbly(input, serial);

  // Handle tool call with human-in-the-loop approval
  if (hasToolCall && functionCallItem) {
    const parsedArgs = JSON.parse(functionArgs);

    const decision = await requestApproval(
      functionCallItem.callId,
      functionCallItem.name,
      parsedArgs,
    );

    let toolResult: Record<string, unknown>;
    if (decision === 'approved') {
      toolResult = executeTool(functionCallItem.name, parsedArgs);
    } else {
      toolResult = { error: 'The user rejected this action' };
    }

    // Continue the conversation with the tool result
    const followUpInput: OpenAI.Responses.ResponseInput = [
      { role: 'user', content: text },
      {
        type: 'function_call',
        id: functionCallItem.id,
        call_id: functionCallItem.callId,
        name: functionCallItem.name,
        arguments: functionArgs,
      },
      {
        type: 'function_call_output',
        call_id: functionCallItem.callId,
        output: JSON.stringify(toolResult),
      },
    ];

    // Stream the follow-up response, appending to the same message
    channel.appendMessage({ serial, data: '\n\n' });
    await streamToAbly(followUpInput, serial);
  }

  // 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) => {
  const promptId = message.extras?.headers?.promptId;
  if (!promptId) return;

  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();