feat(server): MCP server with actions as tools

package.json

@@ -90,5 +90,8 @@
     },
     "engines": {
         "node": ">=18.0.0 || >=20.0.0"
+    },
+    "dependencies": {
+        "@modelcontextprotocol/sdk": "^1.11.0"
     }
 }

packages/server/lib/controllers/config.controller.ts

@@ -6,10 +6,10 @@ import {
     connectionService,
     errorManager,
     flowService,
-    getActionsByProviderConfigKey,
     getGlobalWebhookReceiveUrl,
     getProvider,
     getProviders,
+    getSimplifiedActionsByProviderConfigKey,
     getSyncConfigsAsStandardConfig,
     getUniqueSyncsByProviderConfig
 } from '@nangohq/shared';
@@ -172,7 +172,7 @@ class ConfigController {
                 };
             });
 
-            const actions = await getActionsByProviderConfigKey(environmentId, providerConfigKey);
+            const actions = await getSimplifiedActionsByProviderConfigKey(environmentId, providerConfigKey);
             const hasWebhook = provider.webhook_routing_script;
             let webhookUrl: string | null = null;
             if (hasWebhook) {

packages/server/lib/controllers/mcp/mcp.ts

@@ -0,0 +1,76 @@
+import { StreamableHTTPServerTransport } from '@modelcontextprotocol/sdk/server/streamableHttp.js';
+import { z } from 'zod';
+
+import { connectionService } from '@nangohq/shared';
+import { zodErrorToHTTP } from '@nangohq/utils';
+
+import { createMcpServerForConnection } from './server.js';
+import { connectionIdSchema, providerConfigKeySchema } from '../../helpers/validation.js';
+import { asyncWrapper } from '../../utils/asyncWrapper.js';
+
+import type { Transport } from '@modelcontextprotocol/sdk/shared/transport.js';
+import type { GetMcp, PostMcp } from '@nangohq/types';
+
+export const validationHeaders = z
+    .object({
+        'connection-id': connectionIdSchema,
+        'provider-config-key': providerConfigKeySchema
+    })
+    .strict();
+
+export const postMcp = asyncWrapper<PostMcp>(async (req, res) => {
+    const valHeaders = validationHeaders.safeParse({ 'connection-id': req.get('connection-id'), 'provider-config-key': req.get('provider-config-key') });
+    if (!valHeaders.success) {
+        res.status(400).send({ error: { code: 'invalid_headers', errors: zodErrorToHTTP(valHeaders.error) } });
+        return;
+    }
+
+    const { environment, account } = res.locals;
+    const headers: PostMcp['Headers'] = valHeaders.data;
+
+    const connectionId = headers['connection-id'];
+    const providerConfigKey = headers['provider-config-key'];
+
+    const { error, response: connection } = await connectionService.getConnection(connectionId, providerConfigKey, environment.id);
+
+    if (error || !connection) {
+        res.status(400).send({
+            error: { code: 'unknown_connection', message: 'Provided connection-id and provider-config-key do not match a valid connection' }
+        });
+        return;
+    }
+
+    const result = await createMcpServerForConnection(account, environment, connection, providerConfigKey);
+    if (result.isErr()) {
+        res.status(500).send({ error: { code: 'Internal server error', message: result.error.message } });
+        return;
+    }
+
+    const server = result.value;
+    const transport: StreamableHTTPServerTransport = new StreamableHTTPServerTransport({
+        sessionIdGenerator: undefined
+    });
+
+    res.on('close', () => {
+        void transport.close();
+        void server.close();
+    });
+
+    // Casting because 'exactOptionalPropertyTypes: true' says `?: string` is not equal to `string | undefined`
+    await server.connect(transport as Transport);
+    await transport.handleRequest(req, res, req.body);
+});
+
+// We have to be explicit about not supporting SSE
+export const getMcp = asyncWrapper<GetMcp>((_, res) => {
+    res.writeHead(405).end(
+        JSON.stringify({
+            jsonrpc: '2.0',
+            error: {
+                code: -32000,
+                message: 'Method not allowed.'
+            },
+            id: null
+        })
+    );
+});

packages/server/lib/controllers/mcp/server.ts

@@ -0,0 +1,158 @@
+import { Server } from '@modelcontextprotocol/sdk/server/index.js';
+import { CallToolRequestSchema, ListToolsRequestSchema } from '@modelcontextprotocol/sdk/types';
+import tracer from 'dd-trace';
+
+import { OtlpSpan, defaultOperationExpiration, logContextGetter } from '@nangohq/logs';
+import { configService, getActionsByProviderConfigKey } from '@nangohq/shared';
+import { Err, Ok, truncateJson } from '@nangohq/utils';
+
+import { getOrchestrator } from '../../utils/utils.js';
+
+import type { CallToolRequest, CallToolResult, Tool } from '@modelcontextprotocol/sdk/types';
+import type { Config } from '@nangohq/shared';
+import type { DBConnectionDecrypted, DBEnvironment, DBSyncConfig, DBTeam, Result } from '@nangohq/types';
+import type { Span } from 'dd-trace';
+import type { JSONSchema7 } from 'json-schema';
+
+export async function createMcpServerForConnection(
+    account: DBTeam,
+    environment: DBEnvironment,
+    connection: DBConnectionDecrypted,
+    providerConfigKey: string
+): Promise<Result<Server>> {
+    const server = new Server(
+        {
+            name: 'Nango MCP server',
+            version: '1.0.0'
+        },
+        {
+            capabilities: {
+                tools: {}
+            }
+        }
+    );
+
+    const providerConfig = await configService.getProviderConfig(providerConfigKey, environment.id);
+
+    if (!providerConfig) {
+        return Err(new Error(`Provider config ${providerConfigKey} not found`));
+    }
+
+    const actions = await getActionsForProvider(environment, providerConfig);
+
+    server.setRequestHandler(ListToolsRequestSchema, () => {
+        return {
+            tools: actions.flatMap((action) => {
+                const tool = actionToTool(action);
+                return tool ? [tool] : [];
+            })
+        };
+    });
+
+    server.setRequestHandler(CallToolRequestSchema, callToolRequestHandler(actions, account, environment, connection, providerConfig));
+
+    return Ok(server);
+}
+
+async function getActionsForProvider(environment: DBEnvironment, providerConfig: Config): Promise<DBSyncConfig[]> {
+    return getActionsByProviderConfigKey(environment.id, providerConfig.unique_key);
+}
+
+function actionToTool(action: DBSyncConfig): Tool | null {
+    const inputSchema =
+        action.input && action.models_json_schema?.definitions && action.models_json_schema?.definitions?.[action.input]
+            ? (action.models_json_schema.definitions[action.input] as JSONSchema7)
+            : ({ type: 'object' } as JSONSchema7);
+
+    if (inputSchema.type !== 'object') {
+        // Invalid input schema, skip this action
+        return null;
+    }
+
+    const description = action.metadata.description || action.sync_name;
+
+    return {
+        name: action.sync_name,
+        inputSchema: {
+            type: 'object',
+            properties: inputSchema.properties,
+            required: inputSchema.required
+        },
+        description
+    };
+}
+
+function callToolRequestHandler(
+    actions: DBSyncConfig[],
+    account: DBTeam,
+    environment: DBEnvironment,
+    connection: DBConnectionDecrypted,
+    providerConfig: Config
+): (request: CallToolRequest) => Promise<CallToolResult> {
+    return async (request: CallToolRequest) => {
+        const active = tracer.scope().active();
+        const span = tracer.startSpan('server.mcp.triggerAction', {
+            childOf: active as Span
+        });
+
+        const { name, arguments: toolArguments } = request.params;
+
+        const action = actions.find((action) => action.sync_name === name);
+
+        if (!action) {
+            span.finish();
+            throw new Error(`Action ${name} not found`);
+        }
+
+        const input = toolArguments ?? {};
+
+        span.setTag('nango.actionName', action.sync_name)
+            .setTag('nango.connectionId', connection.id)
+            .setTag('nango.environmentId', environment.id)
+            .setTag('nango.providerConfigKey', providerConfig.unique_key);
+
+        const logCtx = await logContextGetter.create(
+            { operation: { type: 'action', action: 'run' }, expiresAt: defaultOperationExpiration.action() },
+            {
+                account,
+                environment,
+                integration: { id: providerConfig.id!, name: providerConfig.unique_key, provider: providerConfig.provider },
+                connection: { id: connection.id, name: connection.connection_id },
+                syncConfig: { id: action.id, name: action.sync_name },
+                meta: truncateJson({ input })
+            }
+        );
+        logCtx.attachSpan(new OtlpSpan(logCtx.operation));
+
+        const actionResponse = await getOrchestrator().triggerAction({
+            accountId: account.id,
+            connection,
+            actionName: action.sync_name,
+            input,
+            async: false,
+            retryMax: 3,
+            logCtx
+        });
+
+        if (actionResponse.isOk()) {
+            if (!('data' in actionResponse.value)) {
+                // Shouldn't happen with sync actions.
+                return {
+                    content: []
+                };
+            }
+
+            return {
+                content: [
+                    {
+                        type: 'text',
+                        text: JSON.stringify(actionResponse.value.data, null, 2)
+                    }
+                ]
+            };
+        } else {
+            span.setTag('nango.error', actionResponse.error);
+            throw new Error(actionResponse.error.message);
+        }
+    };
+}

packages/server/lib/routes.public.ts

@@ -5,6 +5,7 @@ import multer from 'multer';
 
 import { connectUrl, flagEnforceCLIVersion } from '@nangohq/utils';
 
+import { getAsyncActionResult } from './controllers/action/getAsyncActionResult.js';
 import appAuthController from './controllers/appAuth.controller.js';
 import { postPublicApiKeyAuthorization } from './controllers/auth/postApiKey.js';
 import { postPublicAppStoreAuthorization } from './controllers/auth/postAppStore.js';
@@ -37,6 +38,7 @@ import { postPublicIntegration } from './controllers/integrations/postIntegratio
 import { deletePublicIntegration } from './controllers/integrations/uniqueKey/deleteIntegration.js';
 import { getPublicIntegration } from './controllers/integrations/uniqueKey/getIntegration.js';
 import { patchPublicIntegration } from './controllers/integrations/uniqueKey/patchIntegration.js';
+import { getMcp, postMcp } from './controllers/mcp/mcp.js';
 import oauthController from './controllers/oauth.controller.js';
 import providerController from './controllers/provider.controller.js';
 import { getPublicProvider } from './controllers/providers/getProvider.js';
@@ -61,7 +63,6 @@ import { resourceCapping } from './middleware/resource-capping.middleware.js';
 import { isBinaryContentType } from './utils/utils.js';
 
 import type { Request, RequestHandler } from 'express';
-import { getAsyncActionResult } from './controllers/action/getAsyncActionResult.js';
 
 const apiAuth: RequestHandler[] = [authMiddleware.secretKeyAuth.bind(authMiddleware), rateLimiterMiddleware];
 const connectSessionAuth: RequestHandler[] = [authMiddleware.connectSessionAuth.bind(authMiddleware), rateLimiterMiddleware];
@@ -218,6 +219,10 @@ publicAPI.route('/sync/status').get(apiAuth, syncController.getSyncStatus.bind(s
 publicAPI.route('/sync/:name/variant/:variant').post(apiAuth, postSyncVariant);
 publicAPI.route('/sync/:name/variant/:variant').delete(apiAuth, deleteSyncVariant);
 
+publicAPI.use('/mcp', jsonContentTypeMiddleware);
+publicAPI.route('/mcp').post(apiAuth, postMcp);
+publicAPI.route('/mcp').get(apiAuth, getMcp);
+
 publicAPI.use('/flow', jsonContentTypeMiddleware);
 publicAPI.route('/flow/attributes').get(apiAuth, syncController.getFlowAttributes.bind(syncController));
 // @deprecated use /scripts/configs

packages/server/package.json

@@ -23,6 +23,8 @@
         "npm": ">=6.14.11"
     },
     "dependencies": {
+        "@modelcontextprotocol/sdk": "^1.11.2",
+        "@nangohq/billing": "file:../billing",
         "@nangohq/database": "file:../database",
         "@nangohq/fleet": "file:../fleet",
         "@nangohq/keystore": "file:../keystore",
@@ -34,7 +36,6 @@
         "@nangohq/shared": "file:../shared",
         "@nangohq/utils": "file:../utils",
         "@nangohq/webhooks": "file:../webhooks",
-        "@nangohq/billing": "file:../billing",
         "@workos-inc/node": "6.2.0",
         "axios": "1.9.0",
         "body-parser": "1.20.3",

packages/shared/lib/services/sync/config/config.service.ts

@@ -236,7 +236,29 @@ export async function getActionConfigByNameAndProviderConfigKey(environment_id:
     return false;
 }
 
-export async function getActionsByProviderConfigKey(environment_id: number, unique_key: string): Promise<Action[]> {
+export async function getActionsByProviderConfigKey(environment_id: number, unique_key: string): Promise<DBSyncConfig[]> {
+    const nango_config_id = await configService.getIdByProviderConfigKey(environment_id, unique_key);
+
+    if (!nango_config_id) {
+        return [];
+    }
+
+    const result = await schema().from<DBSyncConfig>(TABLE).where({
+        environment_id,
+        nango_config_id,
+        deleted: false,
+        active: true,
+        type: 'action'
+    });
+
+    if (result) {
+        return result;
+    }
+
+    return [];
+}
+
+export async function getSimplifiedActionsByProviderConfigKey(environment_id: number, unique_key: string): Promise<Action[]> {
     const nango_config_id = await configService.getIdByProviderConfigKey(environment_id, unique_key);
 
     if (!nango_config_id) {

packages/types/lib/index.ts

@@ -75,3 +75,5 @@ export type * from './fleet/index.js';
 
 export type * from './persist/api.js';
 export type * from './jobs/api.js';
+
+export type * from './mcp/api.js';

packages/types/lib/mcp/api.ts

@@ -0,0 +1,19 @@
+import type { ApiError, Endpoint } from '../api.js';
+
+export type PostMcp = Endpoint<{
+    Method: 'POST';
+    Path: '/mcp';
+    Body: Record<string, unknown>;
+    Headers: {
+        'connection-id': string;
+        'provider-config-key': string;
+    };
+    Success: Record<string, unknown>;
+    Error: ApiError<'missing_connection_id' | 'unknown_connection'>;
+}>;
+
+export type GetMcp = Endpoint<{
+    Method: 'GET';
+    Path: '/mcp';
+    Success: Record<string, unknown>;
+}>;

packages/types/lib/syncConfigs/db.ts

@@ -1,7 +1,7 @@
-import type { JSONSchema7 } from 'json-schema';
 import type { TimestampsAndDeleted } from '../db';
 import type { LegacySyncModelSchema, NangoConfigMetadata } from '../deploy/incomingFlow';
 import type { NangoModel, ScriptTypeLiteral, SyncTypeLiteral } from '../nangoYaml';
+import type { JSONSchema7 } from 'json-schema';
 
 export interface DBSyncConfig extends TimestampsAndDeleted {
     id: number;