import "dotenv/config";
import { openai } from "@ai-sdk/openai";
import { Composio } from "@composio/core";
import * as readline from "readline";
import { streamText, type ModelMessage, stepCountIs } from "ai";
import { createMCPClient } from "@ai-sdk/mcp";
const composioAPIKey = process.env.COMPOSIO_API_KEY;
const composioUserID = process.env.COMPOSIO_USER_ID;
if (!process.env.OPENAI_API_KEY) throw new Error("OPENAI_API_KEY is not set");
if (!composioAPIKey) throw new Error("COMPOSIO_API_KEY is not set");
if (!composioUserID) throw new Error("COMPOSIO_USER_ID is not set");
const composio = new Composio({
apiKey: composioAPIKey,
});
async function main() {
const session = await composio.create(composioUserID!, {
toolkits: ["ably"],
});
const mcpUrl = session.mcp.url;
const mcpClient = await createMCPClient({
transport: {
type: "http",
url: mcpUrl,
headers: session.mcp.headers,
},
});
const tools = await mcpClient.tools();
let messages: ModelMessage[] = [];
console.log("Chat started! Type 'exit' or 'quit' to end the conversation.\n");
console.log(
"Ask any questions related to ably, like summarize my last 5 emails, send an email, etc... :)))\n",
);
const rl = readline.createInterface({
input: process.stdin,
output: process.stdout,
prompt: "> ",
});
rl.prompt();
rl.on("line", async (userInput: string) => {
const trimmedInput = userInput.trim();
if (["exit", "quit", "bye"].includes(trimmedInput.toLowerCase())) {
console.log("\nGoodbye!");
rl.close();
process.exit(0);
}
if (!trimmedInput) {
rl.prompt();
return;
}
messages.push({ role: "user", content: trimmedInput });
console.log("\nAgent is thinking...\n");
try {
const stream = streamText({
model: openai("gpt-5"),
messages,
tools,
toolChoice: "auto",
stopWhen: stepCountIs(10),
onStepFinish: (step) => {
for (const toolCall of step.toolCalls) {
console.log(`[Using tool: ${toolCall.toolName}]`);
}
if (step.toolCalls.length > 0) {
console.log("");
}
},
});
for await (const chunk of stream.textStream) {
process.stdout.write(chunk);
}
console.log("\n\n---\n");
const response = await stream.response;
if (response?.messages?.length) {
messages.push(...response.messages);
}
} catch (error) {
console.error("\nAn error occurred while talking to the agent:");
console.error(error);
console.log(
"\nYou can try again or restart the app if it keeps happening.\n",
);
} finally {
rl.prompt();
}
});
rl.on("close", async () => {
await mcpClient.close();
console.log("\n👋 Session ended.");
process.exit(0);
});
}
main().catch((err) => {
console.error("Fatal error:", err);
process.exit(1);
});
