Setup Arcade with LangChain

ARCADE_API_KEY=YOUR_ARCADE_API_KEY
OPENAI_API_KEY=YOUR_OPENAI_API_KEY

"use strict";
import { Arcade } from "@arcadeai/arcadejs";
import {
  type ToolExecuteFunctionFactoryInput,
  executeZodTool,
  isAuthorizationRequiredError,
  toZod,
} from "@arcadeai/arcadejs/lib";
import { type ToolExecuteFunction } from "@arcadeai/arcadejs/lib/zod/types";
import { createAgent, tool } from "langchain";
import {
  Command,
  interrupt,
  MemorySaver,
  type Interrupt,
} from "@langchain/langgraph";
import chalk from "chalk";
import readline from "node:readline/promises";

// configure your own values to customize your agent
 
// The Arcade User ID identifies who is authorizing each service.
const arcadeUserID = "{arcade_user_id}";
// This determines which MCP server is providing the tools, you can customize this to make a Notion agent. All tools from the MCP servers defined in the array will be used.
const MCPServers = ["Slack"];
// This determines individual tools. Useful to pick specific tools when you don't need all of them.
const individualTools = ["Gmail_ListEmails", "Gmail_SendEmail", "Gmail_WhoAmI"];
// This determines the maximum number of tool definitions Arcade will return per MCP server
const toolLimit = 30;
// This prompt defines the behavior of the agent.
const systemPrompt =
  "You are a helpful assistant that can use Gmail tools. Your main task is to help the user with anything they may need.";
// This determines which LLM will be used inside the agent
const agentModel = "gpt-4o-mini";
// This allows LangChain to retain the context of the session
const threadID = "1";

function executeOrInterruptTool({
  zodToolSchema,
  toolDefinition,
  client,
  userId,
}: ToolExecuteFunctionFactoryInput): ToolExecuteFunction<any> {
  const { name: toolName } = zodToolSchema;
 
  return async (input: unknown) => {
    try {
      // Try to execute the tool
      const result = await executeZodTool({
        zodToolSchema,
        toolDefinition,
        client,
        userId,
      })(input);
      return result;
    } catch (error) {
      // If the tool requires authorization, interrupt the flow and ask the user to authorize the tool
      if (error instanceof Error && isAuthorizationRequiredError(error)) {
        const response = await client.tools.authorize({
          tool_name: toolName,
          user_id: userId,
        });
 
        // Interrupt the flow here, and pass everything the handler needs to get the user's authorization
        const interrupt_response = interrupt({
          authorization_required: true,
          authorization_response: response,
          tool_name: toolName,
          url: response.url ?? "",
        });
 
        // If the user authorized the tool, retry the tool call without interrupting the flow
        if (interrupt_response.authorized) {
          const result = await executeZodTool({
            zodToolSchema,
            toolDefinition,
            client,
            userId,
          })(input);
          return result;
        } else {
          // If the user didn't authorize the tool, throw an error, which will be handled by LangChain
          throw new Error(
            `Authorization required for tool call ${toolName}, but user didn't authorize.`
          );
        }
      }
      throw error;
    }
  };
}

// Initialize the Arcade client
const arcade = new Arcade();
 
export type GetToolsProps = {
  arcade: Arcade;
  mcpServers?: string[];
  individualTools?: string[];
  userId: string;
  limit?: number;
};
 
export async function getTools({
  arcade,
  mcpServers = [],
  individualTools = [],
  userId,
  limit = 30,
}: GetToolsProps) {
  if (mcpServers.length === 0 && individualTools.length === 0) {
    throw new Error("At least one tool or toolkit must be provided");
  }
 
  // Get up to `limit` tools from each configured MCP server
  const from_mcpServers = await Promise.all(
    mcpServers.map(async (mcpServerName) => {
      const definitions = await arcade.tools.list({
        toolkit: mcpServerName,
        limit: limit,
      });
      return definitions.items;
    })
  );
 
  // Get the individual tools
  const from_individualTools = await Promise.all(
    individualTools.map(async (toolName) => {
      return await arcade.tools.get(toolName);
    })
  );
 
  // Combine the tools from the MCP servers and the individual tools
  const all_tools = [...from_mcpServers.flat(), ...from_individualTools];
  const unique_tools = Array.from(
    new Map(all_tools.map((tool) => [tool.qualified_name, tool])).values()
  );
 
  // Convert the Arcade tools to Zod tools
  const arcadeTools = toZod({
    tools: unique_tools,
    client: arcade,
    executeFactory: executeOrInterruptTool,
    userId: userId,
  });
 
  // Convert Arcade tools to LangGraph tools
  const langchainTools = arcadeTools.map(
    ({ name, description, execute, parameters }) =>
      (tool as Function)(execute, {
        name,
        description,
        schema: parameters,
      })
  );
 
  return langchainTools;
}
 
const tools = await getTools({
  arcade,
  mcpServers: MCPServers,
  individualTools: individualTools,
  userId: arcadeUserID,
  limit: toolLimit,
});

async function handleAuthInterrupt(
  interrupt: Interrupt
): Promise<{ authorized: boolean }> {
  const value = interrupt.value;
  const authorization_required = value.authorization_required;
  if (authorization_required) {
    const tool_name = value.tool_name;
    const authorization_response = value.authorization_response;
    console.log("⚙️: Authorization required for tool call", tool_name);
    console.log(
      "⚙️: Please authorize in your browser",
      authorization_response.url
    );
    console.log("⚙️: Waiting for you to complete authorization...");
    try {
      await arcade.auth.waitForCompletion(authorization_response.id);
      console.log("⚙️: Authorization granted. Resuming execution...");
      return { authorized: true };
    } catch (error) {
      console.error("⚙️: Error waiting for authorization to complete:", error);
      return { authorized: false };
    }
  }
  return { authorized: false };
}

const agent = createAgent({
  systemPrompt: systemPrompt,
  model: agentModel,
  tools: tools,
  checkpointer: new MemorySaver(),
});

async function streamAgent(
  agent: any,
  input: any,
  config: any
): Promise<Interrupt[]> {
  const stream = await agent.stream(input, {
    ...config,
    streamMode: "updates",
  });
  const interrupts: Interrupt[] = [];
 
  for await (const chunk of stream) {
    if (chunk.__interrupt__) {
      interrupts.push(...(chunk.__interrupt__ as Interrupt[]));
      continue;
    }
    for (const update of Object.values(chunk)) {
      for (const msg of (update as any)?.messages ?? []) {
        console.log("🤖: ", msg.toFormattedString());
      }
    }
  }
 
  return interrupts;
}

async function main() {
  const config = { configurable: { thread_id: threadID } };
  const rl = readline.createInterface({
    input: process.stdin,
    output: process.stdout,
  });
 
  console.log(chalk.green("Welcome to the chatbot! Type 'exit' to quit."));
  while (true) {
    const input = await rl.question("> ");
    if (input.toLowerCase() === "exit") {
      break;
    }
    rl.pause();
 
    try {
      let agentInput: any = {
        messages: [{ role: "user", content: input }],
      };
 
      // Loop until no more interrupts
      while (true) {
        const interrupts = await streamAgent(agent, agentInput, config);
 
        if (interrupts.length === 0) {
          break; // No more interrupts, we're done
        }
 
        // Handle all interrupts
        const decisions: any[] = [];
        for (const interrupt of interrupts) {
          decisions.push(await handleAuthInterrupt(interrupt, rl));
        }
 
        // Resume with decisions, then loop to check for more interrupts
        // Pass single decision directly, or array for multiple interrupts
        agentInput = new Command({
          resume: decisions.length === 1 ? decisions[0] : decisions,
        });
      }
    } catch (error) {
      console.error(error);
    }
 
    rl.resume();
  }
  console.log(chalk.red("👋 Bye..."));
  process.exit(0);
}
 
// Run the main function
main().catch((err) => console.error(err));