feat: Connections v0.1 proxy /v2/connections to Composio (#190)

Author: lourou

.env.example

@@ -37,6 +37,11 @@ AGENT_POOL_URL=
 AGENT_POOL_API_KEY=
 AGENT_ASSETS_API_KEY=
 
+# Composio (optional — /v2/connections/* returns 503 if unset)
+# Auth configs are looked up dynamically from Composio by toolkit slug.
+COMPOSIO_API_KEY=
+COMPOSIO_CONNECTION_CALLBACK_URL=convos://connections/callback
+
 # S3 Lifecycle Test Configuration
 # Bucket for lifecycle testing (24h expiry policy)
 LIFECYCLE_TEST_BUCKET=

bun.lock

@@ -29,6 +29,7 @@
     "@aws-sdk/client-s3": "^3.750.0",
     "@aws-sdk/s3-request-presigner": "^3.750.0",
     "@bufbuild/protobuf": "^2.2.3",
+    "@composio/core": "^0.6.10",
     "@connectrpc/connect": "^2.0.1",
     "@connectrpc/connect-node": "^2.0.1",
     "@opentelemetry/exporter-trace-otlp-grpc": "^0.200.0",
@@ -58,7 +59,7 @@
     "uint8array-extras": "^1.4.0",
     "uuid": "^11.0.5",
     "viem": "^2.23.2",
-    "zod": "^3.24.1"
+    "zod": "^3.25.76"
   },
   "devDependencies": {
     "@bufbuild/buf": "^1.50.0",

package.json

@@ -0,0 +1,141 @@
+import {
+  Composio,
+  type ConnectedAccountListResponseItem,
+} from "@composio/core";
+import { COMPOSIO_API_KEY, COMPOSIO_CONNECTION_CALLBACK_URL } from "@/config";
+import logger from "@/utils/logger";
+
+// Auth configs rarely change in Composio; cache the toolkit→authConfigId
+// resolution in-process so most initiate calls skip the extra round trip.
+const AUTH_CONFIG_CACHE_TTL_MS = 5 * 60 * 1000;
+
+type AuthConfigCacheEntry = { authConfigId: string | null; expiresAt: number };
+
+export class ComposioService {
+  private composio: Composio;
+  private authConfigCache = new Map<string, AuthConfigCacheEntry>();
+
+  constructor(args: { composio: Composio }) {
+    this.composio = args.composio;
+  }
+
+  /**
+   * Resolve a toolkit slug (e.g. "googlecalendar") to its Composio authConfigId
+   * by querying Composio. Picks the first ENABLED auth config for that toolkit.
+   * Returns null if no enabled config exists.
+   *
+   * Clients must send Composio's canonical slug (case-insensitive).
+   */
+  async resolveAuthConfigId(toolkit: string): Promise<string | null> {
+    const now = Date.now();
+    const normalized = toolkit.toLowerCase();
+    const hit = this.authConfigCache.get(normalized);
+    if (hit && hit.expiresAt > now) {
+      return hit.authConfigId;
+    }
+
+    const list = await this.composio.authConfigs.list({ toolkit: normalized });
+    const enabled = list.items.find(
+      (item) =>
+        item.toolkit.slug.toLowerCase() === normalized &&
+        item.status === "ENABLED",
+    );
+    const authConfigId = enabled?.id ?? null;
+
+    if (!authConfigId) {
+      logger.warn(
+        {
+          toolkit: normalized,
+          returnedItems: list.items.map((item) => ({
+            id: item.id,
+            slug: item.toolkit.slug,
+            status: item.status,
+            isComposioManaged: item.isComposioManaged,
+          })),
+          totalPages: list.totalPages,
+        },
+        "[Composio] resolveAuthConfigId: no ENABLED config matched",
+      );
+    } else {
+      logger.info(
+        { toolkit: normalized, authConfigId },
+        "[Composio] resolveAuthConfigId: resolved",
+      );
+    }
+
+    this.authConfigCache.set(normalized, {
+      authConfigId,
+      expiresAt: now + AUTH_CONFIG_CACHE_TTL_MS,
+    });
+    return authConfigId;
+  }
+
+  async initiate(args: {
+    userId: string;
+    authConfigId: string;
+    callbackUrl?: string;
+  }) {
+    return this.composio.connectedAccounts.initiate(
+      args.userId,
+      args.authConfigId,
+      {
+        callbackUrl: args.callbackUrl ?? COMPOSIO_CONNECTION_CALLBACK_URL,
+      },
+    );
+  }
+
+  /**
+   * Fetch a connected account only if it belongs to the given userId.
+   * Returns null if it doesn't exist or isn't owned by the caller.
+   *
+   * Composio's retrieve endpoint no longer returns userId on the response,
+   * so ownership is verified by listing the caller's accounts.
+   */
+  async getIfOwned(args: { connectionId: string; userId: string }) {
+    const list = await this.composio.connectedAccounts.list({
+      userIds: [args.userId],
+    });
+    const items: ConnectedAccountListResponseItem[] = list.items;
+    return items.find((item) => item.id === args.connectionId) ?? null;
+  }
+
+  async listForUser(userId: string) {
+    return this.composio.connectedAccounts.list({ userIds: [userId] });
+  }
+
+  async delete(connectionId: string) {
+    return this.composio.connectedAccounts.delete(connectionId);
+  }
+}
+
+let cached: ComposioService | null = null;
+let initialized = false;
+
+export function createComposioService(): ComposioService | null {
+  if (initialized) {
+    return cached;
+  }
+  initialized = true;
+
+  if (!COMPOSIO_API_KEY) {
+    logger.warn("[Composio] COMPOSIO_API_KEY not set, connections disabled");
+    return null;
+  }
+
+  cached = new ComposioService({
+    composio: new Composio({
+      apiKey: COMPOSIO_API_KEY,
+      allowTracking: false,
+    }),
+  });
+  logger.info("[Composio] Initialised service");
+  return cached;
+}
+
+// Exposed for tests — resets the singleton.
+export function __resetComposioServiceForTests(
+  override: ComposioService | null = null,
+) {
+  cached = override;
+  initialized = override !== null;
+}

src/api/v2/connections/composio.service.ts

@@ -0,0 +1,12 @@
+import { Router } from "express";
+import { completeHandler } from "./handlers/complete";
+import { deleteHandler } from "./handlers/delete";
+import { initiateHandler } from "./handlers/initiate";
+import { listHandler } from "./handlers/list";
+
+export const connectionsRouter = Router();
+
+connectionsRouter.post("/initiate", initiateHandler);
+connectionsRouter.post("/complete", completeHandler);
+connectionsRouter.get("/", listHandler);
+connectionsRouter.delete("/:id", deleteHandler);

src/api/v2/connections/connections.router.ts

@@ -0,0 +1,57 @@
+import type { Request, Response } from "express";
+import { z } from "zod";
+import { createComposioService } from "../composio.service";
+import { mapComposioToResponse } from "../types";
+
+const bodySchema = z.object({
+  connectionRequestId: z.string().min(1).max(256),
+});
+
+export async function completeHandler(req: Request, res: Response) {
+  const deviceId = res.locals.deviceId;
+  if (!deviceId) {
+    res.status(401).json({ error: "Unauthorized" });
+    return;
+  }
+
+  const parsed = bodySchema.safeParse(req.body);
+  if (!parsed.success) {
+    req.log.warn(
+      {
+        deviceId,
+        issues: parsed.error.issues,
+        receivedBody: req.body as unknown,
+        contentType: req.header("content-type"),
+      },
+      "[Composio] complete body validation failed",
+    );
+    res.status(400).json({
+      error: "Invalid request body",
+      details: parsed.error.issues,
+    });
+    return;
+  }
+
+  const service = createComposioService();
+  if (!service) {
+    res.status(503).json({ error: "Connections not configured" });
+    return;
+  }
+
+  try {
+    const owned = await service.getIfOwned({
+      connectionId: parsed.data.connectionRequestId,
+      userId: deviceId,
+    });
+    if (!owned) {
+      res.status(403).json({ error: "Connection not owned by this device" });
+      return;
+    }
+    res.status(200).json(mapComposioToResponse(owned, deviceId));
+    return;
+  } catch (error) {
+    req.log.error({ error, deviceId }, "[Composio] complete failed");
+    res.status(502).json({ error: "Failed to complete connection" });
+    return;
+  }
+}

src/api/v2/connections/handlers/complete.ts

@@ -0,0 +1,43 @@
+import type { Request, Response } from "express";
+import { createComposioService } from "../composio.service";
+
+export async function deleteHandler(req: Request, res: Response) {
+  const deviceId = res.locals.deviceId;
+  if (!deviceId) {
+    res.status(401).json({ error: "Unauthorized" });
+    return;
+  }
+
+  const connectionId = req.params.id;
+  if (!connectionId) {
+    res.status(400).json({ error: "Missing connection id" });
+    return;
+  }
+
+  const service = createComposioService();
+  if (!service) {
+    res.status(503).json({ error: "Connections not configured" });
+    return;
+  }
+
+  try {
+    const owned = await service.getIfOwned({
+      connectionId,
+      userId: deviceId,
+    });
+    if (!owned) {
+      res.status(403).json({ error: "Connection not owned by this device" });
+      return;
+    }
+    await service.delete(connectionId);
+    res.status(204).send();
+    return;
+  } catch (error) {
+    req.log.error(
+      { error, deviceId, connectionId },
+      "[Composio] delete failed",
+    );
+    res.status(502).json({ error: "Failed to delete connection" });
+    return;
+  }
+}