diff --git a/packages/server/api/src/app/chat/chat-service.ts b/packages/server/api/src/app/chat/chat-service.ts index 38f51794cca..bbefddc0b1d 100644 --- a/packages/server/api/src/app/chat/chat-service.ts +++ b/packages/server/api/src/app/chat/chat-service.ts @@ -29,7 +29,7 @@ import { paginationHelper } from '../helper/pagination/pagination-utils' import { Order } from '../helper/pagination/paginator' import { system } from '../helper/system/system' import { AppSystemProp } from '../helper/system/system-props' -import { mcpServerService } from '../mcp/mcp-service' +import { mcpOAuthTokenService } from '../mcp/oauth/token/mcp-oauth-token.service' import { projectService } from '../project/project-service' import { chatCompaction } from './chat-compaction' import { ChatConversationEntity } from './chat-conversation-entity' @@ -120,7 +120,7 @@ export const chatService = (log: FastifyBaseLogger) => ({ const [conversation, providerConfig, mcpCredentials, projectName, userContent] = await Promise.all([ this.getConversationOrThrow({ id: conversationId, projectId, userId }), resolveChatProvider({ platformId, log }), - getMcpCredentials({ projectId, log }), + getMcpCredentials({ platformId, userId, log }), projectService(log).getOneOrThrow(projectId).then((p) => p.displayName), buildUserContentWithFiles({ text: content, files }), ]) @@ -315,17 +315,18 @@ async function connectMcpClient({ mcpCredentials, log }: { return { mcpClient: client, mcpToolSet } } -async function getMcpCredentials({ projectId, log }: { projectId: string, log: FastifyBaseLogger }): Promise<{ mcpServerUrl: string | null, mcpToken: string | null }> { - const { data: mcpServer, error } = await tryCatch(async () => mcpServerService(log).getByProjectId(projectId)) +async function getMcpCredentials({ platformId, userId, log }: { platformId: string, userId: string, log: FastifyBaseLogger }): Promise<{ mcpServerUrl: string | null, mcpToken: string | null }> { + const { data: accessToken, error } = await tryCatch(() => + mcpOAuthTokenService.issueInternalAccessToken({ userId, platformId, projectId: null }), + ) if (error) { - log.warn({ err: error, projectId }, 'Failed to get MCP credentials — chat will work without MCP tools') + log.warn({ err: error, platformId }, 'Failed to get MCP credentials — chat will work without MCP tools') return { mcpServerUrl: null, mcpToken: null } } const frontendUrl = system.getOrThrow(AppSystemProp.FRONTEND_URL) - const mcpServerUrl = `${frontendUrl}/mcp` return { - mcpServerUrl, - mcpToken: mcpServer.token, + mcpServerUrl: `${frontendUrl}/mcp/platform`, + mcpToken: accessToken, } } diff --git a/packages/server/api/src/app/database/migration/postgres/1788000000000-AddPlatformMcpServer.ts b/packages/server/api/src/app/database/migration/postgres/1788000000000-AddPlatformMcpServer.ts new file mode 100644 index 00000000000..86aad112caa --- /dev/null +++ b/packages/server/api/src/app/database/migration/postgres/1788000000000-AddPlatformMcpServer.ts @@ -0,0 +1,74 @@ +import { QueryRunner } from 'typeorm' +import { Migration } from '../../migration' + +export class AddPlatformMcpServer1788000000000 implements Migration { + name = 'AddPlatformMcpServer1788000000000' + breaking = false + release = '0.82.1' + transaction = true + + public async up(queryRunner: QueryRunner): Promise { + await queryRunner.query(` + ALTER TABLE "mcp_server" + ADD COLUMN "platformId" varchar(21), + ADD COLUMN "type" varchar + `) + + await queryRunner.query(` + UPDATE "mcp_server" + SET "type" = CASE + WHEN "projectId" IS NOT NULL THEN 'PROJECT' + ELSE 'PLATFORM' + END + `) + + await queryRunner.query(` + ALTER TABLE "mcp_server" + ALTER COLUMN "type" SET NOT NULL + `) + + await queryRunner.query(` + ALTER TABLE "mcp_server" + ALTER COLUMN "projectId" DROP NOT NULL + `) + + await queryRunner.query(` + ALTER TABLE "mcp_server" + ADD CONSTRAINT "fk_mcp_server_platform_id" + FOREIGN KEY ("platformId") REFERENCES "platform"("id") + ON DELETE CASCADE ON UPDATE NO ACTION + `) + + await queryRunner.query(` + CREATE UNIQUE INDEX "idx_mcp_server_platform_id" + ON "mcp_server" ("platformId") + `) + } + + public async down(queryRunner: QueryRunner): Promise { + await queryRunner.query(` + DROP INDEX IF EXISTS "idx_mcp_server_platform_id" + `) + + await queryRunner.query(` + ALTER TABLE "mcp_server" + DROP CONSTRAINT IF EXISTS "fk_mcp_server_platform_id" + `) + + // Remove platform MCP servers before making projectId NOT NULL + await queryRunner.query(` + DELETE FROM "mcp_server" WHERE "type" = 'PLATFORM' + `) + + await queryRunner.query(` + ALTER TABLE "mcp_server" + ALTER COLUMN "projectId" SET NOT NULL + `) + + await queryRunner.query(` + ALTER TABLE "mcp_server" + DROP COLUMN "type", + DROP COLUMN "platformId" + `) + } +} diff --git a/packages/server/api/src/app/database/migration/postgres/1789000000000-MakeMcpOAuthProjectIdNullable.ts b/packages/server/api/src/app/database/migration/postgres/1789000000000-MakeMcpOAuthProjectIdNullable.ts new file mode 100644 index 00000000000..1af1404531d --- /dev/null +++ b/packages/server/api/src/app/database/migration/postgres/1789000000000-MakeMcpOAuthProjectIdNullable.ts @@ -0,0 +1,20 @@ +import { QueryRunner } from 'typeorm' +import { Migration } from '../../migration' + +export class MakeMcpOAuthProjectIdNullable1789000000000 implements Migration { + name = 'MakeMcpOAuthProjectIdNullable1789000000000' + release = '0.82.1' + breaking = false + + public async up(queryRunner: QueryRunner): Promise { + await queryRunner.query('ALTER TABLE "mcp_oauth_authorization_code" ALTER COLUMN "projectId" DROP NOT NULL') + await queryRunner.query('ALTER TABLE "mcp_oauth_token" ALTER COLUMN "projectId" DROP NOT NULL') + } + + public async down(queryRunner: QueryRunner): Promise { + await queryRunner.query('DELETE FROM "mcp_oauth_token" WHERE "projectId" IS NULL') + await queryRunner.query('DELETE FROM "mcp_oauth_authorization_code" WHERE "projectId" IS NULL') + await queryRunner.query('ALTER TABLE "mcp_oauth_authorization_code" ALTER COLUMN "projectId" SET NOT NULL') + await queryRunner.query('ALTER TABLE "mcp_oauth_token" ALTER COLUMN "projectId" SET NOT NULL') + } +} diff --git a/packages/server/api/src/app/database/postgres-connection.ts b/packages/server/api/src/app/database/postgres-connection.ts index 12fb574186c..da0ab663a82 100644 --- a/packages/server/api/src/app/database/postgres-connection.ts +++ b/packages/server/api/src/app/database/postgres-connection.ts @@ -368,6 +368,8 @@ import { AddUserSandboxTable1784000000000 } from './migration/postgres/178400000 import { ReplacesSandboxWithVercelAiSdk1785000000000 } from './migration/postgres/1785000000000-ReplacesSandboxWithVercelAiSdk' import { AddChatCompactionColumns1786000000000 } from './migration/postgres/1786000000000-AddChatCompactionColumns' import { AddSsoDomainVerification1787100000000 } from './migration/postgres/1787100000000-AddSsoDomainVerification' +import { AddPlatformMcpServer1788000000000 } from './migration/postgres/1788000000000-AddPlatformMcpServer' +import { MakeMcpOAuthProjectIdNullable1789000000000 } from './migration/postgres/1789000000000-MakeMcpOAuthProjectIdNullable' const getSslConfig = (): boolean | TlsOptions => { const useSsl = system.get(AppSystemProp.POSTGRES_USE_SSL) @@ -751,6 +753,8 @@ export const getMigrations = (): (new () => Migration)[] => { ReplacesSandboxWithVercelAiSdk1785000000000, AddChatCompactionColumns1786000000000, AddSsoDomainVerification1787100000000, + AddPlatformMcpServer1788000000000, + MakeMcpOAuthProjectIdNullable1789000000000, ] return migrations } diff --git a/packages/server/api/src/app/mcp/mcp-entity.ts b/packages/server/api/src/app/mcp/mcp-entity.ts index 7a39c21c9b6..345ced063de 100644 --- a/packages/server/api/src/app/mcp/mcp-entity.ts +++ b/packages/server/api/src/app/mcp/mcp-entity.ts @@ -1,8 +1,9 @@ -import { McpServer, Project } from '@activepieces/shared' +import { McpServer, Platform, Project } from '@activepieces/shared' import { EntitySchema } from 'typeorm' import { ApIdSchema, BaseColumnSchemaPart } from '../database/database-common' -type McpServerWithSchema = McpServer & { +type McpServerWithSchema = McpServer & { + platform: Platform project: Project } @@ -10,7 +11,18 @@ export const McpServerEntity = new EntitySchema({ name: 'mcp_server', columns: { ...BaseColumnSchemaPart, - projectId: ApIdSchema, + platformId: { + ...ApIdSchema, + nullable: true, + }, + projectId: { + ...ApIdSchema, + nullable: true, + }, + type: { + type: String, + nullable: false, + }, status: { type: String, nullable: false, @@ -35,8 +47,23 @@ export const McpServerEntity = new EntitySchema({ columns: ['token'], unique: true, }, + { + name: 'idx_mcp_server_platform_id', + columns: ['platformId'], + unique: true, + }, ], relations: { + platform: { + type: 'many-to-one', + target: 'platform', + cascade: true, + onDelete: 'CASCADE', + joinColumn: { + name: 'platformId', + foreignKeyConstraintName: 'fk_mcp_server_platform_id', + }, + }, project: { type: 'many-to-one', target: 'project', @@ -48,6 +75,4 @@ export const McpServerEntity = new EntitySchema({ }, }, }, - }) - diff --git a/packages/server/api/src/app/mcp/mcp-module.ts b/packages/server/api/src/app/mcp/mcp-module.ts index a8d393d5aec..c671d0f0919 100644 --- a/packages/server/api/src/app/mcp/mcp-module.ts +++ b/packages/server/api/src/app/mcp/mcp-module.ts @@ -1,7 +1,8 @@ import { FastifyPluginAsyncZod } from 'fastify-type-provider-zod' +import { mcpPlatformController } from './mcp-platform-controller' import { mcpServerController } from './mcp-server-controller' export const mcpServerModule: FastifyPluginAsyncZod = async (app) => { await app.register(mcpServerController, { prefix: '/v1/projects/:projectId/mcp-server' }) - + await app.register(mcpPlatformController, { prefix: '/v1/mcp-server' }) } diff --git a/packages/server/api/src/app/mcp/mcp-permissions.ts b/packages/server/api/src/app/mcp/mcp-permissions.ts new file mode 100644 index 00000000000..c57d0ab8d77 --- /dev/null +++ b/packages/server/api/src/app/mcp/mcp-permissions.ts @@ -0,0 +1,69 @@ +import { ActivepiecesError, ApEdition, ErrorCode, isNil, McpToolDefinition, Permission } from '@activepieces/shared' +import { FastifyBaseLogger } from 'fastify' +import { getPrincipalRoleOrThrow } from '../ee/authentication/project-role/rbac-middleware' +import { system } from '../helper/system/system' + +const EDITION_REQUIRES_RBAC = [ApEdition.CLOUD, ApEdition.ENTERPRISE].includes(system.getEdition()) + +export async function resolvePermissionChecker({ userId, projectId, log }: { + userId: string + projectId: string + log: FastifyBaseLogger +}): Promise { + if (!EDITION_REQUIRES_RBAC) { + return ALLOW_ALL + } + + try { + const role = await getPrincipalRoleOrThrow(userId, projectId, log) + const permissionSet = new Set(role.permissions ?? []) + return buildChecker((permission, toolTitle) => { + if (isNil(permission) || permissionSet.has(permission)) { + return null + } + return { + content: [{ type: 'text' as const, text: `❌ Permission denied: your role does not have the "${permission}" permission required to use "${toolTitle}".` }], + isError: true, + } + }) + } + catch (err) { + if (err instanceof ActivepiecesError && err.error.code === ErrorCode.AUTHORIZATION) { + return buildChecker((permission, toolTitle) => { + if (isNil(permission)) { + return null + } + return { + content: [{ type: 'text' as const, text: `❌ Permission denied: no role found for this user in the project. Cannot execute "${toolTitle}".` }], + isError: true, + } + }) + } + throw err + } +} + +export const ALLOW_ALL: PermissionChecker = { + check: () => null, + wrapExecute: ({ execute }) => execute, +} + +function buildChecker(check: PermissionChecker['check']): PermissionChecker { + return { + check, + wrapExecute: ({ execute, permission, toolTitle }) => { + const error = check(permission, toolTitle) + return isNil(error) ? execute : async () => error + }, + } +} + +export type PermissionChecker = { + check: (permission: Permission | undefined, toolTitle: string) => McpToolErrorResult | null + wrapExecute: (params: { execute: McpToolDefinition['execute'], permission: Permission | undefined, toolTitle: string }) => McpToolDefinition['execute'] +} + +type McpToolErrorResult = { + content: Array<{ type: 'text', text: string }> + isError: boolean +} diff --git a/packages/server/api/src/app/mcp/mcp-platform-controller.ts b/packages/server/api/src/app/mcp/mcp-platform-controller.ts new file mode 100644 index 00000000000..11d80456340 --- /dev/null +++ b/packages/server/api/src/app/mcp/mcp-platform-controller.ts @@ -0,0 +1,60 @@ +import { PrincipalType, SERVICE_KEY_SECURITY_OPENAPI, UpdateMcpServerRequest } from '@activepieces/shared' +import { FastifyPluginAsyncZod } from 'fastify-type-provider-zod' +import { securityAccess } from '../core/security/authorization/fastify-security' +import { mcpServerService } from './mcp-service' + +export const mcpPlatformController: FastifyPluginAsyncZod = async (app) => { + + app.get('/', GetPlatformMcpRoute, async (req) => { + return mcpServerService(req.log).getByPlatformId(req.principal.platform.id) + }) + + app.post('/', UpdatePlatformMcpRoute, async (req) => { + const { status, enabledTools } = req.body + return mcpServerService(req.log).updatePlatform({ + platformId: req.principal.platform.id, + status, + enabledTools, + }) + }) + + app.post('/rotate', RotatePlatformTokenRoute, async (req) => { + return mcpServerService(req.log).rotatePlatformToken({ + platformId: req.principal.platform.id, + }) + }) +} + +const GetPlatformMcpRoute = { + config: { + security: securityAccess.platformAdminOnly([PrincipalType.USER]), + }, + schema: { + tags: ['mcp'], + description: 'Get the platform MCP server configuration', + security: [SERVICE_KEY_SECURITY_OPENAPI], + }, +} + +const UpdatePlatformMcpRoute = { + config: { + security: securityAccess.platformAdminOnly([PrincipalType.USER]), + }, + schema: { + tags: ['mcp'], + description: 'Update the platform MCP server configuration', + security: [SERVICE_KEY_SECURITY_OPENAPI], + body: UpdateMcpServerRequest, + }, +} + +const RotatePlatformTokenRoute = { + config: { + security: securityAccess.platformAdminOnly([PrincipalType.USER]), + }, + schema: { + tags: ['mcp'], + description: 'Rotate the platform MCP server token', + security: [SERVICE_KEY_SECURITY_OPENAPI], + }, +} diff --git a/packages/server/api/src/app/mcp/mcp-project-selection.ts b/packages/server/api/src/app/mcp/mcp-project-selection.ts new file mode 100644 index 00000000000..7309ef54061 --- /dev/null +++ b/packages/server/api/src/app/mcp/mcp-project-selection.ts @@ -0,0 +1,17 @@ +const selections = new Map() + +function makeKey({ platformId, userId }: { platformId: string, userId: string }): string { + return `${platformId}:${userId}` +} + +export const mcpProjectSelection = { + get({ platformId, userId }: { platformId: string, userId: string }): string | null { + return selections.get(makeKey({ platformId, userId })) ?? null + }, + set({ platformId, userId, projectId }: { platformId: string, userId: string, projectId: string }): void { + selections.set(makeKey({ platformId, userId }), projectId) + }, + clear({ platformId, userId }: { platformId: string, userId: string }): void { + selections.delete(makeKey({ platformId, userId })) + }, +} diff --git a/packages/server/api/src/app/mcp/mcp-server-builder.ts b/packages/server/api/src/app/mcp/mcp-server-builder.ts new file mode 100644 index 00000000000..b67bd081d94 --- /dev/null +++ b/packages/server/api/src/app/mcp/mcp-server-builder.ts @@ -0,0 +1,244 @@ +import { FlowStatus, isNil, McpProperty, McpPropertyType, mcpToolNameUtils, McpTrigger, Permission, PopulatedMcpServer, ProjectScopedMcpServer, TelemetryEventName } from '@activepieces/shared' +import { McpServer, ResourceTemplate } from '@modelcontextprotocol/sdk/server/mcp.js' +import { FastifyBaseLogger } from 'fastify' +import { z } from 'zod' +import { rejectedPromiseHandler } from '../helper/promise-handler' +import { telemetry } from '../helper/telemetry.utils' +import { WebhookFlowVersionToRun, webhookService } from '../webhooks/webhook.service' +import { ALLOW_ALL, PermissionChecker, resolvePermissionChecker } from './mcp-permissions' +import { mcpProjectSelection } from './mcp-project-selection' +import { activepiecesTools, ALL_CONTROLLABLE_TOOL_NAMES, LOCKED_TOOL_NAMES } from './tools' +import { apSetProjectContextTool } from './tools/ap-set-project-context' + +const MCP_SERVER_INSTRUCTIONS = `## Activepieces MCP Server + +### Workflow +1. Discover: ap_list_pieces, ap_list_connections, ap_list_ai_models +2. Schema: ap_get_piece_props (get field names/types before configuring) +3. Build: ap_build_flow (one call for new flows) OR ap_create_flow → ap_update_trigger → ap_add_step (granular) +4. Validate: ap_validate_flow +5. Publish: ap_lock_and_publish → ap_change_flow_status + +### Key patterns +- **Auth**: ap_list_connections → get \`externalId\` → pass as \`auth\` param on ap_update_step/ap_update_trigger. +- **Step refs**: \`{{stepName.field}}\` — no \`.output.\` in the path (e.g. \`{{trigger.body.email}}\`, \`{{step_1.id}}\`). +- **Step names**: \`trigger\`, \`step_1\`, \`step_2\`, etc. Use ap_flow_structure to see all names. +- **Piece names**: full format (e.g. "@activepieces/piece-slack") for ap_add_step/ap_update_trigger. Short names work for lookup tools. +- **Modifying steps**: use ap_update_step/ap_update_trigger. Never delete+recreate — loses sample data. +- **CODE steps**: export a \`code\` fn; access inputs via \`inputs.key\`. +- **Tables**: use field names, not IDs.` + +export async function buildMcpServer({ mcp, userId, log, resolveProjectMcp }: { + mcp: PopulatedMcpServer + userId: string | null + log: FastifyBaseLogger + resolveProjectMcp?: (projectId: string) => Promise +}): Promise { + const projectId = mcp.projectId + + const server = new McpServer({ + name: 'Activepieces', + title: 'Activepieces', + version: '1.0.0', + websiteUrl: 'https://activepieces.com', + description: 'Automation and workflow MCP server by Activepieces', + icons: [ + { + src: 'https://cdn.activepieces.com/brand/logo.svg', + mimeType: 'image/svg+xml', + }, + { + src: 'https://cdn.activepieces.com/brand/logo-192.png', + mimeType: 'image/png', + sizes: ['192x192'], + }, + ], + }, { + instructions: MCP_SERVER_INSTRUCTIONS, + }) + + if (projectId) { + const permissionChecker = userId + ? await resolvePermissionChecker({ userId, projectId, log }) + : ALLOW_ALL + registerFlowTools({ server, mcp, projectId, permissionChecker, log }) + registerStaticTools({ server, mcp, projectId, permissionChecker, log }) + } + else if (!isNil(mcp.platformId) && !isNil(userId) && !isNil(resolveProjectMcp)) { + registerPlatformTools({ server, mcp, platformId: mcp.platformId, userId, resolveProjectMcp, log }) + } + else { + registerPlaceholderTools(server) + } + + registerEmptyResourcesAndPrompts(server) + return server +} + +function registerPlatformTools({ server, mcp, platformId, userId, resolveProjectMcp, log }: { + server: McpServer + mcp: PopulatedMcpServer + platformId: string + userId: string + resolveProjectMcp: (projectId: string) => Promise + log: FastifyBaseLogger +}): void { + const contextTool = apSetProjectContextTool({ platformId, userId, log }) + server.registerTool(contextTool.title, { + title: contextTool.title, + description: contextTool.description, + inputSchema: contextTool.inputSchema, + annotations: contextTool.annotations, + }, (args: Record) => contextTool.execute(args)) + + const templateMcp: ProjectScopedMcpServer = { ...mcp, projectId: platformId } + const allTools = activepiecesTools(templateMcp, log) + const enabledControllable = new Set(mcp.enabledTools ?? ALL_CONTROLLABLE_TOOL_NAMES) + const tools = allTools.filter(t => LOCKED_TOOL_NAMES.includes(t.title) || enabledControllable.has(t.title)) + + tools.forEach((tool) => { + server.registerTool(tool.title, { + title: tool.title, + description: tool.description, + inputSchema: tool.inputSchema, + annotations: tool.annotations, + }, async (args: Record) => { + const selectedProjectId = mcpProjectSelection.get({ platformId, userId }) + if (isNil(selectedProjectId)) { + return { + content: [{ + type: 'text' as const, + text: 'No project selected. Use ap_set_project_context to select a project first.', + }], + } + } + const projectMcp = await resolveProjectMcp(selectedProjectId) + const projectScopedMcp: ProjectScopedMcpServer = { ...projectMcp, projectId: selectedProjectId } + const permissionChecker = await resolvePermissionChecker({ userId, projectId: selectedProjectId, log }) + const realTools = activepiecesTools(projectScopedMcp, log) + const realTool = realTools.find(t => t.title === tool.title) + if (isNil(realTool)) { + return { + content: [{ type: 'text' as const, text: `Tool "${tool.title}" is not available for this project.` }], + } + } + const execute = permissionChecker.wrapExecute({ execute: realTool.execute, permission: realTool.permission, toolTitle: realTool.title }) + return execute(args) + }) + }) +} + +function registerFlowTools({ server, mcp, projectId, permissionChecker, log }: RegisterToolsParams): void { + const enabledFlows = mcp.flows.filter((flow) => flow.status === FlowStatus.ENABLED) + for (const flow of enabledFlows) { + const mcpTrigger = flow.version.trigger.settings as McpTrigger + const mcpInputs = mcpTrigger.input?.inputSchema ?? [] + const zodFromInputSchema = Object.fromEntries(mcpInputs.map((property) => [property.name, mcpPropertyToZod(property)])) + + const baseName = (mcpTrigger.input?.toolName ?? flow.version.displayName) + '_' + flow.id.substring(0, 4) + const toolName = mcpToolNameUtils.createToolName(baseName) + const toolDescription: string = mcpTrigger.input?.toolDescription ?? '' + + const flowPermissionError = permissionChecker.check(Permission.WRITE_RUN, toolName) + server.registerTool(toolName, { title: toolName, description: toolDescription, inputSchema: zodFromInputSchema }, async (args: Record) => { + if (flowPermissionError) { + return flowPermissionError + } + + const returnsResponse = mcpTrigger.input?.returnsResponse + const response = await webhookService.handleWebhook({ + data: () => Promise.resolve({ + body: {}, + method: 'POST', + headers: {}, + queryParams: {}, + }), + logger: log, + flowId: flow.id, + async: !returnsResponse, + flowVersionToRun: WebhookFlowVersionToRun.LOCKED_FALL_BACK_TO_LATEST, + saveSampleData: false, + payload: args, + execute: true, + failParentOnFailure: false, + }) + const isOkay = Math.floor(response.status / 100) === 2 + + rejectedPromiseHandler(telemetry(log).trackProject(projectId, { + name: TelemetryEventName.MCP_TOOL_CALLED, + payload: { mcpId: projectId, toolName }, + }), log) + + const text = isOkay + ? `✅ Successfully executed flow ${flow.version.displayName}\n\nOutput:\n\`\`\`json\n${JSON.stringify(response, null, 2)}\n\`\`\`` + : `❌ Error executing flow ${flow.version.displayName}\n\nError details:\n\`\`\`json\n${JSON.stringify(response, null, 2) || 'Unknown error occurred'}\n\`\`\`` + + return { content: [{ type: 'text' as const, text }] } + }) + } +} + +function registerStaticTools({ server, mcp, projectId, permissionChecker, log }: RegisterToolsParams): void { + const allTools = activepiecesTools({ ...mcp, projectId }, log) + const enabledControllable = new Set(mcp.enabledTools ?? ALL_CONTROLLABLE_TOOL_NAMES) + const tools = allTools.filter(t => LOCKED_TOOL_NAMES.includes(t.title) || enabledControllable.has(t.title)) + + tools.forEach((tool) => { + const execute = permissionChecker.wrapExecute({ execute: tool.execute, permission: tool.permission, toolTitle: tool.title }) + server.registerTool(tool.title, { title: tool.title, description: tool.description, inputSchema: tool.inputSchema, annotations: tool.annotations }, (args: Record) => execute(args)) + }) +} + +function registerPlaceholderTools(server: McpServer): void { + const allToolNames = [...LOCKED_TOOL_NAMES, ...ALL_CONTROLLABLE_TOOL_NAMES] + allToolNames.forEach((toolName) => { + server.registerTool(toolName, { + title: toolName, + description: `${toolName} — requires a project to be selected first.`, + }, async () => ({ + content: [{ type: 'text' as const, text: `No project selected. Please select a project from the dropdown in the chat input area before using ${toolName}.` }], + })) + }) +} + +function mcpPropertyToZod(property: McpProperty): z.ZodTypeAny { + const base = (() => { + switch (property.type) { + case McpPropertyType.TEXT: + case McpPropertyType.DATE: + return z.string() + case McpPropertyType.NUMBER: + return z.number() + case McpPropertyType.BOOLEAN: + return z.boolean() + case McpPropertyType.ARRAY: + return z.array(z.string()) + case McpPropertyType.OBJECT: + return z.record(z.string(), z.string()) + default: + return z.unknown() + } + })() + const described = property.description ? base.describe(property.description) : base + return property.required ? described : described.nullish() +} + +function registerEmptyResourcesAndPrompts(server: McpServer): void { + server.registerResource( + '_', + new ResourceTemplate('activepieces://empty', { + list: async () => ({ resources: [] }), + }), + {}, + async () => ({ contents: [] }), + ) + server.registerPrompt('_', {}, () => ({ messages: [] })) +} + +type RegisterToolsParams = { + server: McpServer + mcp: PopulatedMcpServer + projectId: string + permissionChecker: PermissionChecker + log: FastifyBaseLogger +} diff --git a/packages/server/api/src/app/mcp/mcp-server-controller.ts b/packages/server/api/src/app/mcp/mcp-server-controller.ts index 29ae4345a02..a60664859cc 100644 --- a/packages/server/api/src/app/mcp/mcp-server-controller.ts +++ b/packages/server/api/src/app/mcp/mcp-server-controller.ts @@ -12,7 +12,7 @@ export const mcpServerController: FastifyPluginAsyncZod = async (app) => { }) app.post('/', UpdateMcpRequest, async (req) => { - const { status, enabledTools } = req.body as UpdateMcpServerRequest + const { status, enabledTools } = req.body return mcpServerService(req.log).update({ projectId: req.projectId, status, diff --git a/packages/server/api/src/app/mcp/mcp-service.ts b/packages/server/api/src/app/mcp/mcp-service.ts index a0a70ba3c71..fd6f9b3cd63 100644 --- a/packages/server/api/src/app/mcp/mcp-service.ts +++ b/packages/server/api/src/app/mcp/mcp-service.ts @@ -1,223 +1,109 @@ -import { ActivepiecesError, ApEdition, apId, ErrorCode, FlowStatus, FlowTriggerType, FlowVersionState, isNil, MCP_TRIGGER_PIECE_NAME, McpProperty, McpPropertyType, McpServer as McpServerSchema, McpServerStatus, McpToolDefinition, mcpToolNameUtils, McpTrigger, Permission, PopulatedFlow, PopulatedMcpServer, spreadIfNotUndefined, TelemetryEventName } from '@activepieces/shared' -import { McpServer, ResourceTemplate } from '@modelcontextprotocol/sdk/server/mcp.js' +import { apId, FlowTriggerType, FlowVersionState, isNil, MCP_TRIGGER_PIECE_NAME, McpServer as McpServerSchema, McpServerStatus, McpServerType, PopulatedFlow, PopulatedMcpServer, spreadIfNotUndefined, tryCatch } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' -import { z } from 'zod' import { repoFactory } from '../core/db/repo-factory' -import { getPrincipalRoleOrThrow } from '../ee/authentication/project-role/rbac-middleware' import { flowService } from '../flows/flow/flow.service' -import { rejectedPromiseHandler } from '../helper/promise-handler' -import { system } from '../helper/system/system' -import { telemetry } from '../helper/telemetry.utils' -import { WebhookFlowVersionToRun, webhookService } from '../webhooks/webhook.service' import { McpServerEntity } from './mcp-entity' -import { activepiecesTools, ALL_CONTROLLABLE_TOOL_NAMES, LOCKED_TOOL_NAMES } from './tools' - -const EDITION_REQUIRES_RBAC = [ApEdition.CLOUD, ApEdition.ENTERPRISE].includes(system.getEdition()) - -const MCP_SERVER_INSTRUCTIONS = `## Activepieces MCP Server - -### Workflow -1. Discover: ap_list_pieces, ap_list_connections, ap_list_ai_models -2. Schema: ap_get_piece_props (get field names/types before configuring) -3. Build: ap_build_flow (one call for new flows) OR ap_create_flow → ap_update_trigger → ap_add_step (granular) -4. Validate: ap_validate_flow -5. Publish: ap_lock_and_publish → ap_change_flow_status - -### Key patterns -- **Auth**: ap_list_connections → get \`externalId\` → pass as \`auth\` param on ap_update_step/ap_update_trigger. -- **Step refs**: \`{{stepName.field}}\` — no \`.output.\` in the path (e.g. \`{{trigger.body.email}}\`, \`{{step_1.id}}\`). -- **Step names**: \`trigger\`, \`step_1\`, \`step_2\`, etc. Use ap_flow_structure to see all names. -- **Piece names**: full format (e.g. "@activepieces/piece-slack") for ap_add_step/ap_update_trigger. Short names work for lookup tools. -- **Modifying steps**: use ap_update_step/ap_update_trigger. Never delete+recreate — loses sample data. -- **CODE steps**: export a \`code\` fn; access inputs via \`inputs.key\`. -- **Tables**: use field names, not IDs.` +import { buildMcpServer } from './mcp-server-builder' +import { ALL_CONTROLLABLE_TOOL_NAMES } from './tools' export const mcpServerRepository = repoFactory(McpServerEntity) -export const mcpServerService = (log: FastifyBaseLogger) => { - return { - getPopulatedByProjectId: async (projectId: string): Promise => { - const mcp = await mcpServerService(log).getByProjectId(projectId) - const flows = await listFlows(mcp, log) - return { - ...mcp, - flows, - } - }, - getByProjectId: async (projectId: string): Promise => { - const mcpServer = await mcpServerRepository().findOneBy({ projectId }) - if (isNil(mcpServer)) { - await mcpServerRepository().upsert({ - id: apId(), - status: McpServerStatus.DISABLED, - projectId, - token: apId(72), - enabledTools: ALL_CONTROLLABLE_TOOL_NAMES, - }, ['projectId']) - return mcpServerRepository().findOneByOrFail({ projectId }) - } - return mcpServer - }, - rotateToken: async ({ projectId }: RotateTokenRequest): Promise => { - const mcp = await mcpServerService(log).getByProjectId(projectId) - await mcpServerRepository().update(mcp.id, { - token: apId(72), - }) - return mcpServerService(log).getPopulatedByProjectId(projectId) - }, - update: async ({ projectId, status, enabledTools }: UpdateParams) => { - const mcp = await mcpServerService(log).getByProjectId(projectId) - const patch = { - ...spreadIfNotUndefined('status', status), - ...spreadIfNotUndefined('enabledTools', enabledTools), - } - if (Object.keys(patch).length > 0) { - await mcpServerRepository().update(mcp.id, patch) - } - return mcpServerService(log).getPopulatedByProjectId(projectId) - }, - buildServer: async ({ mcp, userId }: BuildServerRequest): Promise => { - const permissionChecker = await resolvePermissionChecker({ userId, projectId: mcp.projectId, log }) - - const server = new McpServer({ - name: 'Activepieces', - title: 'Activepieces', - version: '1.0.0', - websiteUrl: 'https://activepieces.com', - description: 'Automation and workflow MCP server by Activepieces', - icons: [ - { - src: 'https://cdn.activepieces.com/brand/logo.svg', - mimeType: 'image/svg+xml', - }, - { - src: 'https://cdn.activepieces.com/brand/logo-192.png', - mimeType: 'image/png', - sizes: ['192x192'], - }, - ], - }, { - instructions: MCP_SERVER_INSTRUCTIONS, - }) - const enabledFlows = mcp.flows.filter((flow) => flow.status === FlowStatus.ENABLED) - for (const flow of enabledFlows) { - const mcpTrigger = flow.version.trigger.settings as McpTrigger - const mcpInputs = mcpTrigger.input?.inputSchema ?? [] - const zodFromInputSchema = Object.fromEntries(mcpInputs.map((property) => [property.name, mcpPropertyToZod(property)])) - - const baseName = (mcpTrigger.input?.toolName ?? flow.version.displayName) + '_' + flow.id.substring(0, 4) - const toolName = mcpToolNameUtils.createToolName(baseName) - const toolDescription: string = mcpTrigger.input?.toolDescription ?? '' - - const flowPermissionError = permissionChecker.check(Permission.WRITE_RUN, toolName) - server.tool(toolName, toolDescription, zodFromInputSchema, { title: toolName }, async (args) => { - if (flowPermissionError) { - return flowPermissionError - } - - const returnsResponse = mcpTrigger.input?.returnsResponse - const response = await webhookService.handleWebhook({ - data: () => { - return Promise.resolve({ - body: {}, - method: 'POST', - headers: {}, - queryParams: {}, - }) - }, - logger: log, - flowId: flow.id, - async: !returnsResponse, - flowVersionToRun: WebhookFlowVersionToRun.LOCKED_FALL_BACK_TO_LATEST, - saveSampleData: false, - payload: args, - execute: true, - failParentOnFailure: false, - }) - const isOkay = Math.floor(response.status / 100) === 2 - - rejectedPromiseHandler(telemetry(log).trackProject(mcp.projectId, { - name: TelemetryEventName.MCP_TOOL_CALLED, - payload: { - mcpId: mcp.projectId, - toolName, - }, - }), log) - - if (isOkay) { - return { - content: [{ - type: 'text', - text: `✅ Successfully executed flow ${flow.version.displayName}\n\n` + - `Output:\n\`\`\`json\n${JSON.stringify(response, null, 2)}\n\`\`\``, - }], - } - } - return { - content: [ - { - type: 'text', - text: `❌ Error executing flow ${flow.version.displayName}\n\n` + - `Error details:\n\`\`\`json\n${JSON.stringify(response, null, 2) || 'Unknown error occurred'}\n\`\`\``, - }, - ], - } - }) - } - - const allTools = activepiecesTools(mcp, log) - const enabledControllable = new Set(mcp.enabledTools ?? ALL_CONTROLLABLE_TOOL_NAMES) - const tools = allTools.filter(t => LOCKED_TOOL_NAMES.includes(t.title) || enabledControllable.has(t.title)) - tools.forEach((tool) => { - const execute = permissionChecker.wrapExecute({ execute: tool.execute, permission: tool.permission, toolTitle: tool.title }) - server.registerTool(tool.title, { title: tool.title, description: tool.description, inputSchema: tool.inputSchema, annotations: tool.annotations }, (args: Record) => execute(args)) - }) - - registerEmptyResourcesAndPrompts(server) - return server - }, +export const mcpServerService = (log: FastifyBaseLogger) => ({ + getByProjectId: async (projectId: string): Promise => { + return getOrCreate({ + where: { projectId }, + defaults: { type: McpServerType.PROJECT, status: McpServerStatus.DISABLED, projectId, platformId: null }, + }) + }, + + getByPlatformId: async (platformId: string): Promise => { + return getOrCreate({ + where: { platformId }, + defaults: { type: McpServerType.PLATFORM, status: McpServerStatus.ENABLED, platformId, projectId: null }, + }) + }, + + getPopulatedByProjectId: async (projectId: string): Promise => { + const mcp = await mcpServerService(log).getByProjectId(projectId) + const flows = await listMcpFlows(projectId, log) + return { ...mcp, flows } + }, + + getPopulatedByPlatformId: async (platformId: string): Promise => { + const mcp = await mcpServerService(log).getByPlatformId(platformId) + return { ...mcp, flows: [] } + }, + + rotateToken: async ({ projectId }: { projectId: string }): Promise => { + const mcp = await mcpServerService(log).getByProjectId(projectId) + await mcpServerRepository().update(mcp.id, { token: apId(72) }) + return mcpServerService(log).getPopulatedByProjectId(projectId) + }, + + rotatePlatformToken: async ({ platformId }: { platformId: string }): Promise => { + const mcp = await mcpServerService(log).getByPlatformId(platformId) + await mcpServerRepository().update(mcp.id, { token: apId(72) }) + return mcpServerService(log).getByPlatformId(platformId) + }, + + update: async ({ projectId, status, enabledTools }: UpdateParams): Promise => { + const mcp = await mcpServerService(log).getByProjectId(projectId) + await applyPatch(mcp.id, { status, enabledTools }) + return mcpServerService(log).getPopulatedByProjectId(projectId) + }, + + updatePlatform: async ({ platformId, status, enabledTools }: UpdatePlatformParams): Promise => { + const mcp = await mcpServerService(log).getByPlatformId(platformId) + await applyPatch(mcp.id, { status, enabledTools }) + return mcpServerService(log).getByPlatformId(platformId) + }, + + buildServer: async ({ mcp, userId }: { mcp: PopulatedMcpServer, userId: string | null }) => { + return buildMcpServer({ + mcp, + userId, + log, + resolveProjectMcp: (projectId: string) => mcpServerService(log).getPopulatedByProjectId(projectId), + }) + }, +}) + +async function getOrCreate({ where, defaults }: { + where: { projectId: string } | { platformId: string } + defaults: { type: McpServerType, status: McpServerStatus, projectId: string | null, platformId: string | null } +}): Promise { + const existing = await mcpServerRepository().findOneBy(where) + if (!isNil(existing)) return existing + const { data: created, error } = await tryCatch(async () => + mcpServerRepository().save({ + id: apId(), + ...defaults, + token: apId(72), + enabledTools: ALL_CONTROLLABLE_TOOL_NAMES, + }), + ) + if (error) { + // Unique constraint violation from a concurrent insert — the other request won + const fallback = await mcpServerRepository().findOneBy(where) + if (!isNil(fallback)) return fallback + throw error } + return created } - -export async function resolvePermissionChecker({ userId, projectId, log }: { userId: string, projectId: string, log: FastifyBaseLogger }): Promise { - const allowAll: PermissionChecker = { - check: () => null, - wrapExecute: ({ execute }) => execute, +async function applyPatch(id: string, { status, enabledTools }: { status?: McpServerStatus, enabledTools?: string[] }): Promise { + const patch = { + ...spreadIfNotUndefined('status', status), + ...spreadIfNotUndefined('enabledTools', enabledTools), } - if (!EDITION_REQUIRES_RBAC) { - return allowAll + if (Object.keys(patch).length > 0) { + await mcpServerRepository().update(id, patch) } - - let userPermissions: string[] - try { - const role = await getPrincipalRoleOrThrow(userId, projectId, log) - userPermissions = role.permissions ?? [] - } - catch (err) { - if (err instanceof ActivepiecesError && err.error.code === ErrorCode.AUTHORIZATION) { - return buildChecker((permission, toolTitle) => { - if (isNil(permission)) { - return null - } - return noRoleError(toolTitle) - }) - } - throw err - } - - const permissionSet = new Set(userPermissions) - return buildChecker((permission, toolTitle) => { - if (isNil(permission) || permissionSet.has(permission)) { - return null - } - return missingPermissionError(permission, toolTitle) - }) } -async function listFlows(mcp: McpServerSchema, logger: FastifyBaseLogger): Promise { +async function listMcpFlows(projectId: string, logger: FastifyBaseLogger): Promise { const flows = await flowService(logger).list({ - projectIds: [mcp.projectId], + projectIds: [projectId], limit: 1000000, cursorRequest: null, versionState: FlowVersionState.DRAFT, @@ -226,97 +112,14 @@ async function listFlows(mcp: McpServerSchema, logger: FastifyBaseLogger): Promi return flows.data.filter((flow) => flow.version.trigger.type === FlowTriggerType.PIECE && flow.version.trigger.settings.pieceName === MCP_TRIGGER_PIECE_NAME) } -function mcpPropertyToZod(property: McpProperty): z.ZodTypeAny { - let schema: z.ZodTypeAny - - switch (property.type) { - case McpPropertyType.TEXT: - case McpPropertyType.DATE: - schema = z.string() - break - case McpPropertyType.NUMBER: - schema = z.number() - break - case McpPropertyType.BOOLEAN: - schema = z.boolean() - break - case McpPropertyType.ARRAY: - schema = z.array(z.string()) - break - case McpPropertyType.OBJECT: - schema = z.record(z.string(), z.string()) - break - default: - schema = z.unknown() - } - - if (property.description) { - schema = schema.describe(property.description) - } - - return property.required ? schema : schema.nullish() -} - -function registerEmptyResourcesAndPrompts(server: McpServer): void { - server.registerResource( - '_', - new ResourceTemplate('activepieces://empty', { - list: async () => ({ resources: [] }), - }), - {}, - async () => ({ contents: [] }), - ) - server.registerPrompt('_', {}, () => ({ messages: [] })) -} - -function buildChecker(check: PermissionChecker['check']): PermissionChecker { - return { - check, - wrapExecute: ({ execute, permission, toolTitle }) => { - const error = check(permission, toolTitle) - if (isNil(error)) { - return execute - } - return async () => error - }, - } -} - -function noRoleError(toolTitle: string): McpToolErrorResult { - return { - content: [{ type: 'text' as const, text: `❌ Permission denied: no role found for this user in the project. Cannot execute "${toolTitle}".` }], - isError: true, - } -} - -function missingPermissionError(permission: Permission, toolTitle: string): McpToolErrorResult { - return { - content: [{ type: 'text' as const, text: `❌ Permission denied: your role does not have the "${permission}" permission required to use "${toolTitle}".` }], - isError: true, - } -} - -type PermissionChecker = { - check: (permission: Permission | undefined, toolTitle: string) => McpToolErrorResult | null - wrapExecute: (params: { execute: McpToolDefinition['execute'], permission: Permission | undefined, toolTitle: string }) => McpToolDefinition['execute'] -} - -type McpToolErrorResult = { - content: Array<{ type: 'text', text: string }> - isError: boolean -} - -type BuildServerRequest = { - mcp: PopulatedMcpServer - userId: string -} - -type RotateTokenRequest = { +type UpdateParams = { projectId: string + status?: McpServerStatus + enabledTools?: string[] } -type UpdateParams = { +type UpdatePlatformParams = { + platformId: string status?: McpServerStatus - projectId: string enabledTools?: string[] } diff --git a/packages/server/api/src/app/mcp/oauth/code/mcp-oauth-approve.controller.ts b/packages/server/api/src/app/mcp/oauth/code/mcp-oauth-approve.controller.ts index afa48a20e2b..669cdd5a007 100644 --- a/packages/server/api/src/app/mcp/oauth/code/mcp-oauth-approve.controller.ts +++ b/packages/server/api/src/app/mcp/oauth/code/mcp-oauth-approve.controller.ts @@ -1,9 +1,10 @@ -import { Permission, PrincipalType } from '@activepieces/shared' +import { isNil, PlatformRole, PrincipalType } from '@activepieces/shared' import { FastifyPluginAsyncZod } from 'fastify-type-provider-zod' import { z } from 'zod' -import { ProjectResourceType } from '../../../core/security/authorization/common' import { securityAccess } from '../../../core/security/authorization/fastify-security' import { JwtAudience, jwtUtils } from '../../../helper/jwt-utils' +import { projectService } from '../../../project/project-service' +import { userService } from '../../../user/user-service' import { mcpOAuthCodeService } from './mcp-oauth-code.service' export const mcpOAuthApproveController: FastifyPluginAsyncZod = async (app) => { @@ -13,6 +14,24 @@ export const mcpOAuthApproveController: FastifyPluginAsyncZod = async (app) => { const userId = req.principal.id const platformId = req.principal.platform.id + if (isNil(projectId)) { + const user = await userService(req.log).getOneOrFail({ id: userId }) + if (user.platformRole !== PlatformRole.ADMIN) { + return reply.status(403).send({ error: 'access_denied', error_description: 'Only platform administrators can authorize platform-wide MCP access' }) + } + } + else { + const user = await userService(req.log).getOneOrFail({ id: userId }) + const accessibleProjects = await projectService(req.log).getAllForUser({ + platformId, + userId, + isPrivileged: userService(req.log).isUserPrivileged(user), + }) + if (!accessibleProjects.some(p => p.id === projectId)) { + return reply.status(403).send({ error: 'access_denied', error_description: 'You do not have access to this project' }) + } + } + const key = await jwtUtils.getJwtSecret() let authRequest: AuthRequestPayload try { @@ -33,7 +52,7 @@ export const mcpOAuthApproveController: FastifyPluginAsyncZod = async (app) => { const code = await mcpOAuthCodeService.create({ clientId: authRequest.clientId, userId, - projectId, + projectId: projectId ?? null, platformId, redirectUri: authRequest.redirectUri, codeChallenge: authRequest.codeChallenge, @@ -54,17 +73,13 @@ export const mcpOAuthApproveController: FastifyPluginAsyncZod = async (app) => { const ApproveRequest = { config: { - security: securityAccess.project( - [PrincipalType.USER], - Permission.WRITE_MCP, - { type: ProjectResourceType.BODY }, - ), + security: securityAccess.publicPlatform([PrincipalType.USER]), }, schema: { tags: ['mcp-oauth'], body: z.object({ authRequestId: z.string(), - projectId: z.string(), + projectId: z.string().optional(), }), }, } @@ -76,5 +91,6 @@ type AuthRequestPayload = { codeChallengeMethod: string state: string | null scopes: string[] + resource: string | null type: 'mcp_auth_request' } diff --git a/packages/server/api/src/app/mcp/oauth/code/mcp-oauth-authorize.controller.ts b/packages/server/api/src/app/mcp/oauth/code/mcp-oauth-authorize.controller.ts index cf32a2af075..d70de583e48 100644 --- a/packages/server/api/src/app/mcp/oauth/code/mcp-oauth-authorize.controller.ts +++ b/packages/server/api/src/app/mcp/oauth/code/mcp-oauth-authorize.controller.ts @@ -12,7 +12,7 @@ const AUTH_REQUEST_TTL_10_MINUTES_SECONDS = 10 * 60 export const mcpOAuthAuthorizeController: FastifyPluginAsyncZod = async (app) => { app.get('/authorize', AuthorizeRequest, async (req, reply) => { - const { client_id, redirect_uri, response_type, code_challenge, code_challenge_method, state, scope } = req.query + const { client_id, redirect_uri, response_type, code_challenge, code_challenge_method, state, scope, resource } = req.query if (response_type !== 'code') { return reply.status(400).send({ error: 'unsupported_response_type' }) @@ -41,6 +41,7 @@ export const mcpOAuthAuthorizeController: FastifyPluginAsyncZod = async (app) => codeChallengeMethod: code_challenge_method, state: state ?? null, scopes: scope ? scope.split(' ') : ['mcp'], + resource: resource ?? null, type: 'mcp_auth_request', }, key, diff --git a/packages/server/api/src/app/mcp/oauth/code/mcp-oauth-code.entity.ts b/packages/server/api/src/app/mcp/oauth/code/mcp-oauth-code.entity.ts index e60005a46ff..9fda85edde6 100644 --- a/packages/server/api/src/app/mcp/oauth/code/mcp-oauth-code.entity.ts +++ b/packages/server/api/src/app/mcp/oauth/code/mcp-oauth-code.entity.ts @@ -17,7 +17,10 @@ export const McpOAuthAuthorizationCodeEntity = new EntitySchema { + registerMcpEndpoint(app, McpServerType.PROJECT) +} + +export const mcpPlatformHttpController: FastifyPluginAsyncZod = async (app) => { + registerMcpEndpoint(app, McpServerType.PLATFORM) +} +function registerMcpEndpoint(app: Parameters[0], scope: McpServerType): void { app.addContentTypeParser( 'application/json', { parseAs: 'string' }, @@ -18,7 +24,6 @@ export const mcpOAuthHttpController: FastifyPluginAsyncZod = async (app) => { if (body == null || body.trim() === '') { return done(null, {}) } - try { done(null, JSON.parse(body)) } @@ -48,7 +53,7 @@ export const mcpOAuthHttpController: FastifyPluginAsyncZod = async (app) => { }) } - const identity = await resolveIdentity(token, req.log) + const identity = await resolveIdentity({ token, scope, log: req.log }) if (isNil(identity)) { return reply.status(401).send({ error: 'unauthorized', @@ -56,26 +61,15 @@ export const mcpOAuthHttpController: FastifyPluginAsyncZod = async (app) => { }) } - rejectedPromiseHandler(telemetry(req.log).trackProject(identity.projectId, { - name: TelemetryEventName.MCP_SERVER_CONNECTED, - payload: { - projectId: identity.projectId, - userId: identity.userId, - }, - }), req.log) - - let mcp: PopulatedMcpServer - try { - mcp = await mcpServerService(req.log).getPopulatedByProjectId(identity.projectId) - } - catch (err) { - req.log.debug({ err }, 'Failed to resolve MCP server for project') + const { mcp, userId } = await resolveMcpAndUser({ identity, log: req.log }) + if (isNil(mcp)) { return reply.status(401).send({ error: 'unauthorized', message: 'Invalid project or token.', }) } - const { server } = await mcpServerService(req.log).buildServer({ mcp, userId: identity.userId }) + + const { server } = await mcpServerService(req.log).buildServer({ mcp, userId }) const transport = new StreamableHTTPServerTransport({ sessionIdGenerator: undefined, @@ -91,31 +85,56 @@ export const mcpOAuthHttpController: FastifyPluginAsyncZod = async (app) => { }) } -async function resolveIdentity(token: string, log: FastifyBaseLogger): Promise { - if (token.split('.').length === 3) { - try { - const payload = await mcpOAuthTokenService.verifyAccessToken(token) - return { projectId: payload.projectId, userId: payload.sub } - } - catch (e) { - log.debug({ err: e }, 'JWT verification failed') - } +async function resolveIdentity({ token, scope, log }: { token: string, scope: McpServerType, log: FastifyBaseLogger }): Promise { + const { data: payload, error } = await tryCatch(() => mcpOAuthTokenService.verifyAccessToken(token)) + if (error) { + log.debug({ err: error }, 'OAuth token verification failed') + return null } - - const mcpServer = await mcpServerRepository().findOneBy({ token }) - if (!isNil(mcpServer)) { - const project = await projectService(log).getOneOrThrow(mcpServer.projectId) - return { projectId: mcpServer.projectId, userId: project.ownerId } + const { projectId } = payload + const isPlatformToken = isNil(projectId) + if (isPlatformToken && scope === McpServerType.PLATFORM) { + return { type: McpServerType.PLATFORM, platformId: payload.platformId, userId: payload.sub } + } + if (!isPlatformToken && scope === McpServerType.PROJECT) { + return { type: McpServerType.PROJECT, projectId, userId: payload.sub } } - return null } -type ResolvedIdentity = { - projectId: string - userId: string +async function resolveMcpAndUser({ identity, log }: { identity: ResolvedIdentity, log: FastifyBaseLogger }): Promise<{ mcp: PopulatedMcpServer | null, userId: string | null }> { + try { + if (identity.type === McpServerType.PLATFORM) { + rejectedPromiseHandler(telemetry(log).trackPlatform(identity.platformId, { + name: TelemetryEventName.MCP_SERVER_CONNECTED, + payload: { + platformId: identity.platformId, + userId: identity.userId, + }, + }), log) + const mcp = await mcpServerService(log).getPopulatedByPlatformId(identity.platformId) + return { mcp, userId: identity.userId } + } + rejectedPromiseHandler(telemetry(log).trackProject(identity.projectId, { + name: TelemetryEventName.MCP_SERVER_CONNECTED, + payload: { + projectId: identity.projectId, + userId: identity.userId, + }, + }), log) + const mcp = await mcpServerService(log).getPopulatedByProjectId(identity.projectId) + return { mcp, userId: identity.userId } + } + catch (err) { + log.debug({ err }, 'Failed to resolve MCP server') + return { mcp: null, userId: null } + } } +type ResolvedIdentity = + | { type: McpServerType.PROJECT, projectId: string, userId: string } + | { type: McpServerType.PLATFORM, platformId: string, userId: string } + const McpEndpointConfig = { config: { security: securityAccess.public() }, schema: { hide: true }, diff --git a/packages/server/api/src/app/mcp/oauth/metadata/mcp-oauth-metadata.controller.ts b/packages/server/api/src/app/mcp/oauth/metadata/mcp-oauth-metadata.controller.ts index a20b183abd2..89e4ba7914a 100644 --- a/packages/server/api/src/app/mcp/oauth/metadata/mcp-oauth-metadata.controller.ts +++ b/packages/server/api/src/app/mcp/oauth/metadata/mcp-oauth-metadata.controller.ts @@ -27,6 +27,14 @@ export const mcpOAuthMetadataController: FastifyPluginAsyncZod = async (app) => authorization_servers: [issuer], }) }) + + app.get('/.well-known/oauth-protected-resource/mcp/platform', ProtectedResourceMetadataRequest, async (_req, reply) => { + const issuer = mcpOAuthTokenService.getIssuerUrl() + return reply.status(200).header('Access-Control-Allow-Origin', '*').send({ + resource: `${issuer}/mcp/platform`, + authorization_servers: [issuer], + }) + }) } const AuthorizationServerMetadataRequest = { diff --git a/packages/server/api/src/app/mcp/oauth/token/mcp-oauth-token.entity.ts b/packages/server/api/src/app/mcp/oauth/token/mcp-oauth-token.entity.ts index c8ffa765f81..c4b7c2f2ca5 100644 --- a/packages/server/api/src/app/mcp/oauth/token/mcp-oauth-token.entity.ts +++ b/packages/server/api/src/app/mcp/oauth/token/mcp-oauth-token.entity.ts @@ -17,7 +17,10 @@ export const McpOAuthTokenEntity = new EntitySchema({ nullable: false, }, userId: ApIdSchema, - projectId: ApIdSchema, + projectId: { + ...ApIdSchema, + nullable: true, + }, platformId: ApIdSchema, scopes: { type: String, diff --git a/packages/server/api/src/app/mcp/oauth/token/mcp-oauth-token.service.ts b/packages/server/api/src/app/mcp/oauth/token/mcp-oauth-token.service.ts index 38783a9a99e..2597dddaa7c 100644 --- a/packages/server/api/src/app/mcp/oauth/token/mcp-oauth-token.service.ts +++ b/packages/server/api/src/app/mcp/oauth/token/mcp-oauth-token.service.ts @@ -12,6 +12,7 @@ const repo = repoFactory(McpOAuthTokenEntity) const ACCESS_TOKEN_TTL_15_MINUTES_SECONDS = 15 * 60 const REFRESH_TOKEN_TTL_30_DAYS_MS = 30 * 24 * 60 * 60 * 1000 +const INTERNAL_CHAT_CLIENT_ID = 'internal-chat' function generateRefreshToken(): string { return randomBytes(48).toString('base64url') @@ -125,6 +126,10 @@ export const mcpOAuthTokenService = { await repo().update(criteria, { revoked: true }) }, + async issueInternalAccessToken({ userId, platformId, projectId }: { userId: string, platformId: string, projectId: string | null }): Promise { + return issueAccessToken({ userId, platformId, projectId, clientId: INTERNAL_CHAT_CLIENT_ID, scopes: ['mcp'] }) + }, + getIssuerUrl(): string { return system.get(AppSystemProp.MCP_OAUTH_ISSUER_URL) ?? system.getOrThrow(AppSystemProp.FRONTEND_URL) @@ -142,7 +147,7 @@ export class OAuthTokenError extends Error { type IssueAccessTokenParams = { userId: string - projectId: string + projectId: string | null platformId: string clientId: string scopes: string[] @@ -154,7 +159,7 @@ type ExchangeCodeParams = { codeChallengeMethod: string clientId: string userId: string - projectId: string + projectId: string | null platformId: string scopes: string[] } @@ -173,7 +178,7 @@ type TokenResponse = { export type McpOAuthAccessTokenPayload = { sub: string - projectId: string + projectId: string | null platformId: string clientId: string scopes: string[] diff --git a/packages/server/api/src/app/mcp/tools/ap-add-branch.ts b/packages/server/api/src/app/mcp/tools/ap-add-branch.ts index c9dd1fedff7..5bf1112a72a 100644 --- a/packages/server/api/src/app/mcp/tools/ap-add-branch.ts +++ b/packages/server/api/src/app/mcp/tools/ap-add-branch.ts @@ -3,9 +3,9 @@ import { FlowOperationRequest, FlowOperationType, isNil, - McpServer, McpToolDefinition, Permission, + ProjectScopedMcpServer, } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' import { z } from 'zod' @@ -20,7 +20,7 @@ const addBranchInput = z.object({ conditions: mcpUtils.BRANCH_CONDITIONS_INPUT_SCHEMA.optional(), }) -export const apAddBranchTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apAddBranchTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_add_branch', permission: Permission.WRITE_FLOW, diff --git a/packages/server/api/src/app/mcp/tools/ap-add-step.ts b/packages/server/api/src/app/mcp/tools/ap-add-step.ts index 0e815a94cd7..53e963262aa 100644 --- a/packages/server/api/src/app/mcp/tools/ap-add-step.ts +++ b/packages/server/api/src/app/mcp/tools/ap-add-step.ts @@ -5,9 +5,9 @@ import { FlowOperationType, flowStructureUtil, isNil, - McpServer, McpToolDefinition, Permission, + ProjectScopedMcpServer, RouterExecutionType, StepLocationRelativeToParent, UpdateActionRequest, @@ -35,7 +35,7 @@ const addStepInput = z.object({ loopItems: z.string().optional(), }) -export const apAddStepTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apAddStepTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_add_step', permission: Permission.WRITE_FLOW, diff --git a/packages/server/api/src/app/mcp/tools/ap-build-flow.ts b/packages/server/api/src/app/mcp/tools/ap-build-flow.ts index 84eedb43b34..0e2b42d5fc9 100644 --- a/packages/server/api/src/app/mcp/tools/ap-build-flow.ts +++ b/packages/server/api/src/app/mcp/tools/ap-build-flow.ts @@ -4,10 +4,10 @@ import { FlowOperationType, flowStructureUtil, FlowTriggerType, - McpServer, McpToolDefinition, Permission, PieceTrigger, + ProjectScopedMcpServer, RouterExecutionType, StepLocationRelativeToParent, UpdateActionRequest, @@ -43,7 +43,7 @@ const buildFlowInput = z.object({ steps: z.array(stepSpec), }) -export const apBuildFlowTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apBuildFlowTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_build_flow', permission: Permission.WRITE_FLOW, diff --git a/packages/server/api/src/app/mcp/tools/ap-change-flow-status.ts b/packages/server/api/src/app/mcp/tools/ap-change-flow-status.ts index 1ef0c4615bd..ce20030254a 100644 --- a/packages/server/api/src/app/mcp/tools/ap-change-flow-status.ts +++ b/packages/server/api/src/app/mcp/tools/ap-change-flow-status.ts @@ -3,9 +3,9 @@ import { FlowOperationType, FlowStatus, isNil, - McpServer, McpToolDefinition, Permission, + ProjectScopedMcpServer, } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' import { z } from 'zod' @@ -18,7 +18,7 @@ const changeFlowStatusInput = z.object({ status: z.enum(Object.values(FlowStatus) as [FlowStatus, ...FlowStatus[]]), }) -export const apChangeFlowStatusTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apChangeFlowStatusTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_change_flow_status', permission: Permission.UPDATE_FLOW_STATUS, diff --git a/packages/server/api/src/app/mcp/tools/ap-create-flow.ts b/packages/server/api/src/app/mcp/tools/ap-create-flow.ts index b900ca563f1..186851969ee 100644 --- a/packages/server/api/src/app/mcp/tools/ap-create-flow.ts +++ b/packages/server/api/src/app/mcp/tools/ap-create-flow.ts @@ -1,10 +1,10 @@ -import { McpServer, McpToolDefinition, Permission } from '@activepieces/shared' +import { McpToolDefinition, Permission, ProjectScopedMcpServer } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' import { z } from 'zod' import { flowService } from '../../flows/flow/flow.service' import { mcpUtils } from './mcp-utils' -export const apCreateFlowTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apCreateFlowTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_create_flow', permission: Permission.WRITE_FLOW, diff --git a/packages/server/api/src/app/mcp/tools/ap-create-table.ts b/packages/server/api/src/app/mcp/tools/ap-create-table.ts index 216fcced072..25f64b714c2 100644 --- a/packages/server/api/src/app/mcp/tools/ap-create-table.ts +++ b/packages/server/api/src/app/mcp/tools/ap-create-table.ts @@ -1,4 +1,4 @@ -import { apId, FieldType, McpServer, McpToolDefinition, Permission } from '@activepieces/shared' +import { apId, FieldType, McpToolDefinition, Permission, ProjectScopedMcpServer } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' import { z } from 'zod' import { fieldService } from '../../tables/field/field.service' @@ -15,7 +15,7 @@ const createTableInput = z.object({ })).describe('Fields to create. Max 100 fields per table.'), }) -export const apCreateTableTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apCreateTableTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_create_table', permission: Permission.WRITE_TABLE, diff --git a/packages/server/api/src/app/mcp/tools/ap-delete-branch.ts b/packages/server/api/src/app/mcp/tools/ap-delete-branch.ts index 565433c5264..171680c521c 100644 --- a/packages/server/api/src/app/mcp/tools/ap-delete-branch.ts +++ b/packages/server/api/src/app/mcp/tools/ap-delete-branch.ts @@ -2,9 +2,9 @@ import { FlowOperationRequest, FlowOperationType, isNil, - McpServer, McpToolDefinition, Permission, + ProjectScopedMcpServer, } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' import { z } from 'zod' @@ -18,7 +18,7 @@ const deleteBranchInput = z.object({ branchIndex: z.number().int().min(0), }) -export const apDeleteBranchTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apDeleteBranchTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_delete_branch', permission: Permission.WRITE_FLOW, diff --git a/packages/server/api/src/app/mcp/tools/ap-delete-records.ts b/packages/server/api/src/app/mcp/tools/ap-delete-records.ts index 71144c37a7c..31ee586a99f 100644 --- a/packages/server/api/src/app/mcp/tools/ap-delete-records.ts +++ b/packages/server/api/src/app/mcp/tools/ap-delete-records.ts @@ -1,4 +1,4 @@ -import { McpServer, McpToolDefinition, Permission } from '@activepieces/shared' +import { McpToolDefinition, Permission, ProjectScopedMcpServer } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' import { z } from 'zod' import { recordService } from '../../tables/record/record.service' @@ -8,7 +8,7 @@ const deleteRecordsInput = z.object({ recordIds: z.array(z.string()).describe('Array of record IDs to delete. Use ap_find_records to find them.'), }) -export const apDeleteRecordsTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apDeleteRecordsTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_delete_records', permission: Permission.WRITE_TABLE, diff --git a/packages/server/api/src/app/mcp/tools/ap-delete-step.ts b/packages/server/api/src/app/mcp/tools/ap-delete-step.ts index 220dcadb742..f537f40d20a 100644 --- a/packages/server/api/src/app/mcp/tools/ap-delete-step.ts +++ b/packages/server/api/src/app/mcp/tools/ap-delete-step.ts @@ -3,9 +3,9 @@ import { FlowOperationType, flowStructureUtil, isNil, - McpServer, McpToolDefinition, Permission, + ProjectScopedMcpServer, } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' import { z } from 'zod' @@ -18,7 +18,7 @@ const deleteStepInput = z.object({ stepName: z.string(), }) -export const apDeleteStepTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apDeleteStepTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_delete_step', permission: Permission.WRITE_FLOW, diff --git a/packages/server/api/src/app/mcp/tools/ap-delete-table.ts b/packages/server/api/src/app/mcp/tools/ap-delete-table.ts index 53a03aa991d..3564bc01782 100644 --- a/packages/server/api/src/app/mcp/tools/ap-delete-table.ts +++ b/packages/server/api/src/app/mcp/tools/ap-delete-table.ts @@ -1,4 +1,4 @@ -import { McpServer, McpToolDefinition, Permission } from '@activepieces/shared' +import { McpToolDefinition, Permission, ProjectScopedMcpServer } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' import { z } from 'zod' import { tableService } from '../../tables/table/table.service' @@ -8,7 +8,7 @@ const deleteTableInput = z.object({ tableId: z.string().describe('The ID of the table to delete. Use ap_list_tables to find it.'), }) -export const apDeleteTableTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apDeleteTableTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_delete_table', permission: Permission.WRITE_TABLE, diff --git a/packages/server/api/src/app/mcp/tools/ap-duplicate-flow.ts b/packages/server/api/src/app/mcp/tools/ap-duplicate-flow.ts index e36609e671b..5f229fb693e 100644 --- a/packages/server/api/src/app/mcp/tools/ap-duplicate-flow.ts +++ b/packages/server/api/src/app/mcp/tools/ap-duplicate-flow.ts @@ -1,9 +1,9 @@ import { FlowOperationType, isNil, - McpServer, McpToolDefinition, Permission, + ProjectScopedMcpServer, } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' import { z } from 'zod' @@ -16,7 +16,7 @@ const duplicateFlowInput = z.object({ name: z.string().optional(), }) -export const apDuplicateFlowTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apDuplicateFlowTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_duplicate_flow', permission: Permission.WRITE_FLOW, diff --git a/packages/server/api/src/app/mcp/tools/ap-find-records.ts b/packages/server/api/src/app/mcp/tools/ap-find-records.ts index c0a3726c8c2..4a246864d5f 100644 --- a/packages/server/api/src/app/mcp/tools/ap-find-records.ts +++ b/packages/server/api/src/app/mcp/tools/ap-find-records.ts @@ -1,4 +1,4 @@ -import { FilterOperator, McpServer, McpToolDefinition, Permission } from '@activepieces/shared' +import { FilterOperator, McpToolDefinition, Permission, ProjectScopedMcpServer } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' import { z } from 'zod' import { recordService } from '../../tables/record/record.service' @@ -29,7 +29,7 @@ const findRecordsInput = z.object({ limit: z.number().min(1).max(500).optional().describe('Max records to return (default 50, max 500)'), }) -export const apFindRecordsTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apFindRecordsTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_find_records', permission: Permission.READ_TABLE, diff --git a/packages/server/api/src/app/mcp/tools/ap-flow-structure.ts b/packages/server/api/src/app/mcp/tools/ap-flow-structure.ts index fe0e56f7041..1ba78e82292 100644 --- a/packages/server/api/src/app/mcp/tools/ap-flow-structure.ts +++ b/packages/server/api/src/app/mcp/tools/ap-flow-structure.ts @@ -6,10 +6,10 @@ import { flowStructureUtil, FlowTriggerType, isNil, - McpServer, McpToolDefinition, Note, Permission, + ProjectScopedMcpServer, StepLocationRelativeToParent, } from '@activepieces/shared' import type { Step } from '@activepieces/shared' @@ -280,7 +280,7 @@ function formatFlowStructure( return lines.join('\n') } -export const apFlowStructureTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apFlowStructureTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_flow_structure', permission: Permission.READ_FLOW, diff --git a/packages/server/api/src/app/mcp/tools/ap-get-piece-props.ts b/packages/server/api/src/app/mcp/tools/ap-get-piece-props.ts index 296edbb9cb6..f9fcb871289 100644 --- a/packages/server/api/src/app/mcp/tools/ap-get-piece-props.ts +++ b/packages/server/api/src/app/mcp/tools/ap-get-piece-props.ts @@ -6,8 +6,8 @@ import { FlowVersion, isNil, isObject, - McpServer, McpToolDefinition, + ProjectScopedMcpServer, SampleDataFileType, WorkerJobType, } from '@activepieces/shared' @@ -21,7 +21,7 @@ import { projectService } from '../../project/project-service' import { userInteractionWatcher } from '../../workers/user-interaction-watcher' import { mcpUtils, PropSummary } from './mcp-utils' -export const apGetPiecePropsTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apGetPiecePropsTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_get_piece_props', description: 'Get the input property schema for a piece action or trigger. Returns field names, types, required/optional, defaults, and options. Pass auth to resolve dynamic dropdowns and dynamic property sub-fields (e.g. Custom API Call url/body fields).', diff --git a/packages/server/api/src/app/mcp/tools/ap-get-run.ts b/packages/server/api/src/app/mcp/tools/ap-get-run.ts index 22f6bd22903..7bf602d56c0 100644 --- a/packages/server/api/src/app/mcp/tools/ap-get-run.ts +++ b/packages/server/api/src/app/mcp/tools/ap-get-run.ts @@ -1,4 +1,4 @@ -import { McpServer, McpToolDefinition, Permission } from '@activepieces/shared' +import { McpToolDefinition, Permission, ProjectScopedMcpServer } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' import { z } from 'zod' import { flowRunService } from '../../flows/flow-run/flow-run-service' @@ -9,7 +9,7 @@ const getRunInput = z.object({ flowRunId: z.string().describe('The ID of the flow run. Use ap_list_runs to find it.'), }) -export const apGetRunTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apGetRunTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_get_run', permission: Permission.READ_RUN, diff --git a/packages/server/api/src/app/mcp/tools/ap-insert-records.ts b/packages/server/api/src/app/mcp/tools/ap-insert-records.ts index cf116abf893..f18231bdbc8 100644 --- a/packages/server/api/src/app/mcp/tools/ap-insert-records.ts +++ b/packages/server/api/src/app/mcp/tools/ap-insert-records.ts @@ -1,4 +1,4 @@ -import { McpServer, McpToolDefinition, Permission } from '@activepieces/shared' +import { McpToolDefinition, Permission, ProjectScopedMcpServer } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' import { z } from 'zod' import { recordService } from '../../tables/record/record.service' @@ -10,7 +10,7 @@ const insertRecordsInput = z.object({ records: z.array(z.record(z.string(), z.string())).min(1).max(50).describe('Array of records (1–50). Each record maps field names to values. Example: [{"Name": "Alice", "Age": "30"}]'), }) -export const apInsertRecordsTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apInsertRecordsTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_insert_records', permission: Permission.WRITE_TABLE, diff --git a/packages/server/api/src/app/mcp/tools/ap-list-ai-models.ts b/packages/server/api/src/app/mcp/tools/ap-list-ai-models.ts index 22bec2950b3..1c9b688c303 100644 --- a/packages/server/api/src/app/mcp/tools/ap-list-ai-models.ts +++ b/packages/server/api/src/app/mcp/tools/ap-list-ai-models.ts @@ -1,4 +1,4 @@ -import { AIProviderModelType, AIProviderName, McpServer, McpToolDefinition } from '@activepieces/shared' +import { AIProviderModelType, AIProviderName, McpToolDefinition, ProjectScopedMcpServer } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' import { z } from 'zod' import { aiProviderService } from '../../ai/ai-provider-service' @@ -11,7 +11,7 @@ const listAiModelsInput = z.object({ provider: providerSchema.optional().describe('Filter by provider name. Omit to list all configured providers and their models.'), }) -export const apListAiModelsTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apListAiModelsTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_list_ai_models', description: 'List configured AI providers and their available models. Use this to discover valid provider and model values for configuring Run Agent steps. The output shows provider names and model IDs needed for the aiProviderModel input.', diff --git a/packages/server/api/src/app/mcp/tools/ap-list-connections.ts b/packages/server/api/src/app/mcp/tools/ap-list-connections.ts index f4a980d4351..c93862575d4 100644 --- a/packages/server/api/src/app/mcp/tools/ap-list-connections.ts +++ b/packages/server/api/src/app/mcp/tools/ap-list-connections.ts @@ -1,8 +1,8 @@ import { AppConnectionStatus, - McpServer, McpToolDefinition, Permission, + ProjectScopedMcpServer, } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' import { z } from 'zod' @@ -33,7 +33,7 @@ const listConnectionsSchema = z.object({ ), }) -export const apListConnectionsTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apListConnectionsTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_list_connections', permission: Permission.READ_APP_CONNECTION, diff --git a/packages/server/api/src/app/mcp/tools/ap-list-flows.ts b/packages/server/api/src/app/mcp/tools/ap-list-flows.ts index e01d08fd187..1cb840d3d42 100644 --- a/packages/server/api/src/app/mcp/tools/ap-list-flows.ts +++ b/packages/server/api/src/app/mcp/tools/ap-list-flows.ts @@ -1,4 +1,4 @@ -import { FlowStatus, FlowTriggerType, isNil, McpServer, McpToolDefinition, Permission, PopulatedFlow } from '@activepieces/shared' +import { FlowStatus, FlowTriggerType, isNil, McpToolDefinition, Permission, PopulatedFlow, ProjectScopedMcpServer } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' import { z } from 'zod' import { flowService } from '../../flows/flow/flow.service' @@ -13,7 +13,7 @@ const listFlowsInput = z.object({ name: z.string().optional(), }) -export const apListFlowsTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apListFlowsTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_list_flows', permission: Permission.READ_FLOW, diff --git a/packages/server/api/src/app/mcp/tools/ap-list-pieces.ts b/packages/server/api/src/app/mcp/tools/ap-list-pieces.ts index 17b0304e041..45f7224cf2a 100644 --- a/packages/server/api/src/app/mcp/tools/ap-list-pieces.ts +++ b/packages/server/api/src/app/mcp/tools/ap-list-pieces.ts @@ -1,8 +1,8 @@ import { LocalesEnum, - McpServer, McpToolDefinition, PieceCategory, + ProjectScopedMcpServer, SuggestionType, } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' @@ -21,7 +21,7 @@ const listPiecesSchema = z.object({ includeTriggers: z.boolean().optional(), }) -export const apListPiecesTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apListPiecesTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_list_pieces', description: 'List available pieces with their actions and triggers. Use includeActions/includeTriggers for details.', diff --git a/packages/server/api/src/app/mcp/tools/ap-list-runs.ts b/packages/server/api/src/app/mcp/tools/ap-list-runs.ts index 46c29759ee0..b976531d7df 100644 --- a/packages/server/api/src/app/mcp/tools/ap-list-runs.ts +++ b/packages/server/api/src/app/mcp/tools/ap-list-runs.ts @@ -1,4 +1,4 @@ -import { FlowRunStatus, isNil, McpServer, McpToolDefinition, Permission, RunEnvironment } from '@activepieces/shared' +import { FlowRunStatus, isNil, McpToolDefinition, Permission, ProjectScopedMcpServer, RunEnvironment } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' import { z } from 'zod' import { flowRunService } from '../../flows/flow-run/flow-run-service' @@ -15,7 +15,7 @@ const listRunsInput = z.object({ limit: z.number().min(1).max(50).optional().describe('Max runs to return (default 10, max 50)'), }) -export const apListRunsTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apListRunsTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_list_runs', permission: Permission.READ_RUN, diff --git a/packages/server/api/src/app/mcp/tools/ap-list-tables.ts b/packages/server/api/src/app/mcp/tools/ap-list-tables.ts index 836c2f65d44..fed6c94fa15 100644 --- a/packages/server/api/src/app/mcp/tools/ap-list-tables.ts +++ b/packages/server/api/src/app/mcp/tools/ap-list-tables.ts @@ -1,11 +1,11 @@ -import { McpServer, McpToolDefinition, Permission } from '@activepieces/shared' +import { McpToolDefinition, Permission, ProjectScopedMcpServer } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' import { fieldService } from '../../tables/field/field.service' import { tableService } from '../../tables/table/table.service' import { mcpUtils } from './mcp-utils' import { formatFieldInfo } from './table-utils' -export const apListTablesTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apListTablesTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_list_tables', permission: Permission.READ_TABLE, diff --git a/packages/server/api/src/app/mcp/tools/ap-lock-and-publish.ts b/packages/server/api/src/app/mcp/tools/ap-lock-and-publish.ts index 66441b75269..9bbe5c72c87 100644 --- a/packages/server/api/src/app/mcp/tools/ap-lock-and-publish.ts +++ b/packages/server/api/src/app/mcp/tools/ap-lock-and-publish.ts @@ -4,9 +4,9 @@ import { FlowStatus, flowStructureUtil, isNil, - McpServer, McpToolDefinition, Permission, + ProjectScopedMcpServer, } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' import { z } from 'zod' @@ -18,7 +18,7 @@ const lockAndPublishInput = z.object({ flowId: z.string(), }) -export const apLockAndPublishTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apLockAndPublishTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_lock_and_publish', permission: Permission.UPDATE_FLOW_STATUS, diff --git a/packages/server/api/src/app/mcp/tools/ap-manage-fields.ts b/packages/server/api/src/app/mcp/tools/ap-manage-fields.ts index 1caaacde41a..92381f85b26 100644 --- a/packages/server/api/src/app/mcp/tools/ap-manage-fields.ts +++ b/packages/server/api/src/app/mcp/tools/ap-manage-fields.ts @@ -1,4 +1,4 @@ -import { FieldType, isNil, McpServer, McpToolDefinition, Permission } from '@activepieces/shared' +import { FieldType, isNil, McpToolDefinition, Permission, ProjectScopedMcpServer } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' import { z } from 'zod' import { fieldService } from '../../tables/field/field.service' @@ -14,7 +14,7 @@ const manageFieldsInput = z.object({ options: z.array(z.string()).optional().describe('Dropdown options (required for ADD with STATIC_DROPDOWN type)'), }) -export const apManageFieldsTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apManageFieldsTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_manage_fields', permission: Permission.WRITE_TABLE, diff --git a/packages/server/api/src/app/mcp/tools/ap-manage-notes.ts b/packages/server/api/src/app/mcp/tools/ap-manage-notes.ts index d14bed9f929..9f5413f9c4d 100644 --- a/packages/server/api/src/app/mcp/tools/ap-manage-notes.ts +++ b/packages/server/api/src/app/mcp/tools/ap-manage-notes.ts @@ -3,10 +3,10 @@ import { FlowOperationRequest, FlowOperationType, isNil, - McpServer, McpToolDefinition, NoteColorVariant, Permission, + ProjectScopedMcpServer, } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' import { z } from 'zod' @@ -30,7 +30,7 @@ const manageNotesInput = z.object({ }).optional(), }) -export const apManageNotesTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apManageNotesTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_manage_notes', permission: Permission.WRITE_FLOW, diff --git a/packages/server/api/src/app/mcp/tools/ap-rename-flow.ts b/packages/server/api/src/app/mcp/tools/ap-rename-flow.ts index ab3f1e9c2f2..fa8adb6569c 100644 --- a/packages/server/api/src/app/mcp/tools/ap-rename-flow.ts +++ b/packages/server/api/src/app/mcp/tools/ap-rename-flow.ts @@ -2,9 +2,9 @@ import { FlowOperationRequest, FlowOperationType, isNil, - McpServer, McpToolDefinition, Permission, + ProjectScopedMcpServer, } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' import { z } from 'zod' @@ -17,7 +17,7 @@ const renameFlowInput = z.object({ displayName: z.string(), }) -export const apRenameFlowTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apRenameFlowTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_rename_flow', permission: Permission.WRITE_FLOW, diff --git a/packages/server/api/src/app/mcp/tools/ap-retry-run.ts b/packages/server/api/src/app/mcp/tools/ap-retry-run.ts index 33d8d0d2526..ad5970c3e31 100644 --- a/packages/server/api/src/app/mcp/tools/ap-retry-run.ts +++ b/packages/server/api/src/app/mcp/tools/ap-retry-run.ts @@ -1,4 +1,4 @@ -import { FlowRetryStrategy, FlowRunStatus, isFlowRunStateTerminal, isNil, McpServer, McpToolDefinition, Permission } from '@activepieces/shared' +import { FlowRetryStrategy, FlowRunStatus, isFlowRunStateTerminal, isNil, McpToolDefinition, Permission, ProjectScopedMcpServer } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' import { z } from 'zod' import { flowService } from '../../flows/flow/flow.service' @@ -13,7 +13,7 @@ const retryRunInput = z.object({ strategy: z.enum(retryStrategyValues).describe('FROM_FAILED_STEP to resume where it failed, ON_LATEST_VERSION to re-run with the current published flow.'), }) -export const apRetryRunTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apRetryRunTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_retry_run', permission: Permission.WRITE_RUN, diff --git a/packages/server/api/src/app/mcp/tools/ap-run-action.ts b/packages/server/api/src/app/mcp/tools/ap-run-action.ts index 72fb78c9a96..8bd8556e7e8 100644 --- a/packages/server/api/src/app/mcp/tools/ap-run-action.ts +++ b/packages/server/api/src/app/mcp/tools/ap-run-action.ts @@ -1,4 +1,4 @@ -import { McpServer, McpToolDefinition, Permission } from '@activepieces/shared' +import { McpToolDefinition, Permission, ProjectScopedMcpServer } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' import { z } from 'zod' import { executeAdhocAction } from './flow-run-utils' @@ -12,7 +12,7 @@ const runActionInput = z.object({ pieceVersion: z.string().optional().describe('Defaults to the latest installed version. Use "~X.Y.Z" for minor-version pinning.'), }) -export const apRunActionTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apRunActionTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_run_action', permission: Permission.WRITE_RUN, diff --git a/packages/server/api/src/app/mcp/tools/ap-set-project-context.ts b/packages/server/api/src/app/mcp/tools/ap-set-project-context.ts new file mode 100644 index 00000000000..41e2f83c850 --- /dev/null +++ b/packages/server/api/src/app/mcp/tools/ap-set-project-context.ts @@ -0,0 +1,65 @@ +import { isNil, McpToolDefinition } from '@activepieces/shared' +import { FastifyBaseLogger } from 'fastify' +import { z } from 'zod' +import { projectService } from '../../project/project-service' +import { userService } from '../../user/user-service' +import { mcpProjectSelection } from '../mcp-project-selection' + +export const apSetProjectContextTool = ({ platformId, userId, log }: { + platformId: string + userId: string + log: FastifyBaseLogger +}): McpToolDefinition => ({ + title: 'ap_set_project_context', + description: 'Set or clear the active project context. All tools require a project context to operate. Call with a projectId to select a project, or without to clear the selection. Always returns the list of available projects.', + inputSchema: { + projectId: z.string().optional().describe('The project ID to select. Omit to clear the current selection and list available projects.'), + }, + annotations: { + readOnlyHint: false, + idempotentHint: true, + }, + execute: async (args: Record) => { + const projectId = args.projectId as string | undefined + + const user = await userService(log).getOneOrFail({ id: userId }) + const projects = await projectService(log).getAllForUser({ + platformId, + userId, + isPrivileged: userService(log).isUserPrivileged(user), + }) + + if (!isNil(projectId) && projectId !== '') { + const targetProject = projects.find(p => p.id === projectId) + if (!targetProject) { + const projectList = projects.map(p => `- ${p.displayName} (${p.id})`).join('\n') + return { + content: [{ + type: 'text' as const, + text: `Project not found or you don't have access to it.\n\nAvailable projects:\n${projectList}`, + }], + } + } + mcpProjectSelection.set({ platformId, userId, projectId }) + const projectList = projects.map(p => { + const marker = p.id === projectId ? '>' : ' ' + return `${marker} ${p.displayName} (${p.id})` + }).join('\n') + return { + content: [{ + type: 'text' as const, + text: `Project context set to "${targetProject.displayName}".\n\nAll tools will now operate on this project. Use ap_set_project_context without a projectId to clear the selection.\n\nAvailable projects:\n${projectList}`, + }], + } + } + + mcpProjectSelection.clear({ platformId, userId }) + const projectList = projects.map(p => `- ${p.displayName} (${p.id})`).join('\n') + return { + content: [{ + type: 'text' as const, + text: `Project context cleared.\n\nAvailable projects:\n${projectList}\n\nUse ap_set_project_context with a projectId to select one.`, + }], + } + }, +}) diff --git a/packages/server/api/src/app/mcp/tools/ap-setup-guide.ts b/packages/server/api/src/app/mcp/tools/ap-setup-guide.ts index c40ad08c45b..0a4d265ba7f 100644 --- a/packages/server/api/src/app/mcp/tools/ap-setup-guide.ts +++ b/packages/server/api/src/app/mcp/tools/ap-setup-guide.ts @@ -1,5 +1,5 @@ import { PropertyType } from '@activepieces/pieces-framework' -import { AIProviderName, isNil, McpServer, McpToolDefinition } from '@activepieces/shared' +import { AIProviderName, isNil, McpToolDefinition, ProjectScopedMcpServer } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' import { z } from 'zod' import { aiProviderService } from '../../ai/ai-provider-service' @@ -12,7 +12,7 @@ const setupGuideInput = z.object({ pieceName: z.string().optional().describe('For connections: the piece that needs auth (e.g., "@activepieces/piece-gmail"). Omit for general instructions.'), }) -export const apSetupGuideTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apSetupGuideTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_setup_guide', description: 'Get setup instructions for connections or AI providers. Returns steps for the user to follow in the UI.', @@ -35,7 +35,7 @@ export const apSetupGuideTool = (mcp: McpServer, log: FastifyBaseLogger): McpToo } } -async function connectionGuide(mcp: McpServer, log: FastifyBaseLogger, pieceName?: string): Promise<{ content: [{ type: 'text', text: string }] }> { +async function connectionGuide(mcp: ProjectScopedMcpServer, log: FastifyBaseLogger, pieceName?: string): Promise<{ content: [{ type: 'text', text: string }] }> { if (isNil(pieceName)) { return { content: [{ @@ -145,7 +145,7 @@ async function connectionGuide(mcp: McpServer, log: FastifyBaseLogger, pieceName return { content: [{ type: 'text', text: lines.join('\n') }] } } -async function aiProviderGuide(mcp: McpServer, log: FastifyBaseLogger): Promise<{ content: [{ type: 'text', text: string }] }> { +async function aiProviderGuide(mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): Promise<{ content: [{ type: 'text', text: string }] }> { const project = await projectService(log).getOneOrThrow(mcp.projectId) const providers = await aiProviderService(log).listProviders(project.platformId) diff --git a/packages/server/api/src/app/mcp/tools/ap-test-flow.ts b/packages/server/api/src/app/mcp/tools/ap-test-flow.ts index ef2b8cc52eb..47cf27bd375 100644 --- a/packages/server/api/src/app/mcp/tools/ap-test-flow.ts +++ b/packages/server/api/src/app/mcp/tools/ap-test-flow.ts @@ -1,4 +1,4 @@ -import { McpServer, McpToolDefinition, Permission } from '@activepieces/shared' +import { McpToolDefinition, Permission, ProjectScopedMcpServer } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' import { z } from 'zod' import { executeFlowTest } from './flow-run-utils' @@ -9,7 +9,7 @@ const testFlowInput = z.object({ triggerTestData: z.record(z.string(), z.unknown()).optional().describe('Mock trigger output data. Saved as sample data before running the test. Useful when the trigger has no prior test data.'), }) -export const apTestFlowTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apTestFlowTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_test_flow', permission: Permission.WRITE_FLOW, diff --git a/packages/server/api/src/app/mcp/tools/ap-test-step.ts b/packages/server/api/src/app/mcp/tools/ap-test-step.ts index 0da03fcf45c..978090a65e1 100644 --- a/packages/server/api/src/app/mcp/tools/ap-test-step.ts +++ b/packages/server/api/src/app/mcp/tools/ap-test-step.ts @@ -1,4 +1,4 @@ -import { McpServer, McpToolDefinition, Permission } from '@activepieces/shared' +import { McpToolDefinition, Permission, ProjectScopedMcpServer } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' import { z } from 'zod' import { executeFlowTest } from './flow-run-utils' @@ -10,7 +10,7 @@ const testStepInput = z.object({ triggerTestData: z.record(z.string(), z.unknown()).optional().describe('Mock trigger output data. Saved as sample data before running the test. Useful when the trigger has no prior test data.'), }) -export const apTestStepTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apTestStepTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_test_step', permission: Permission.WRITE_FLOW, diff --git a/packages/server/api/src/app/mcp/tools/ap-update-branch.ts b/packages/server/api/src/app/mcp/tools/ap-update-branch.ts index 833ec589beb..b8f67f354e3 100644 --- a/packages/server/api/src/app/mcp/tools/ap-update-branch.ts +++ b/packages/server/api/src/app/mcp/tools/ap-update-branch.ts @@ -5,9 +5,9 @@ import { FlowOperationRequest, FlowOperationType, isNil, - McpServer, McpToolDefinition, Permission, + ProjectScopedMcpServer, RouterActionSettings, } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' @@ -24,7 +24,7 @@ const updateBranchInput = z.object({ conditions: mcpUtils.BRANCH_CONDITIONS_INPUT_SCHEMA.optional(), }) -export const apUpdateBranchTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apUpdateBranchTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_update_branch', permission: Permission.WRITE_FLOW, diff --git a/packages/server/api/src/app/mcp/tools/ap-update-record.ts b/packages/server/api/src/app/mcp/tools/ap-update-record.ts index 9ef2a373cad..02684c2d255 100644 --- a/packages/server/api/src/app/mcp/tools/ap-update-record.ts +++ b/packages/server/api/src/app/mcp/tools/ap-update-record.ts @@ -1,4 +1,4 @@ -import { McpServer, McpToolDefinition, Permission } from '@activepieces/shared' +import { McpToolDefinition, Permission, ProjectScopedMcpServer } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' import { z } from 'zod' import { recordService } from '../../tables/record/record.service' @@ -11,7 +11,7 @@ const updateRecordInput = z.object({ fields: z.record(z.string(), z.string()).describe('Object mapping field names to new values. Only specified fields are updated. Example: {"Name": "Bob", "Age": "25"}'), }) -export const apUpdateRecordTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apUpdateRecordTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_update_record', permission: Permission.WRITE_TABLE, diff --git a/packages/server/api/src/app/mcp/tools/ap-update-step.ts b/packages/server/api/src/app/mcp/tools/ap-update-step.ts index 8c70eaeca32..a9335c04cdf 100644 --- a/packages/server/api/src/app/mcp/tools/ap-update-step.ts +++ b/packages/server/api/src/app/mcp/tools/ap-update-step.ts @@ -4,10 +4,10 @@ import { FlowOperationType, flowStructureUtil, isNil, - McpServer, McpToolDefinition, Permission, PieceActionSettings, + ProjectScopedMcpServer, UpdateActionRequest, } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' @@ -30,7 +30,7 @@ const updateStepInput = z.object({ packageJson: z.string().optional(), }) -export const apUpdateStepTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apUpdateStepTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_update_step', permission: Permission.WRITE_FLOW, diff --git a/packages/server/api/src/app/mcp/tools/ap-update-trigger.ts b/packages/server/api/src/app/mcp/tools/ap-update-trigger.ts index 8fd74e3457a..55f5c8d729c 100644 --- a/packages/server/api/src/app/mcp/tools/ap-update-trigger.ts +++ b/packages/server/api/src/app/mcp/tools/ap-update-trigger.ts @@ -3,10 +3,10 @@ import { FlowOperationType, FlowTriggerType, isNil, - McpServer, McpToolDefinition, Permission, PieceTrigger, + ProjectScopedMcpServer, } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' import { z } from 'zod' @@ -25,7 +25,7 @@ const updateTriggerInput = z.object({ displayName: z.string().optional(), }) -export const apUpdateTriggerTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apUpdateTriggerTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_update_trigger', permission: Permission.WRITE_FLOW, diff --git a/packages/server/api/src/app/mcp/tools/ap-validate-flow.ts b/packages/server/api/src/app/mcp/tools/ap-validate-flow.ts index c85a4204d39..5b9b5b9f2f7 100644 --- a/packages/server/api/src/app/mcp/tools/ap-validate-flow.ts +++ b/packages/server/api/src/app/mcp/tools/ap-validate-flow.ts @@ -3,9 +3,9 @@ import { flowStructureUtil, FlowTriggerType, isNil, - McpServer, McpToolDefinition, Permission, + ProjectScopedMcpServer, Step, } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' @@ -13,7 +13,7 @@ import { z } from 'zod' import { flowService } from '../../flows/flow/flow.service' import { mcpUtils } from './mcp-utils' -export const apValidateFlowTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apValidateFlowTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_validate_flow', permission: Permission.READ_FLOW, diff --git a/packages/server/api/src/app/mcp/tools/ap-validate-step-config.ts b/packages/server/api/src/app/mcp/tools/ap-validate-step-config.ts index cca2bb81f64..c6d3b2cead3 100644 --- a/packages/server/api/src/app/mcp/tools/ap-validate-step-config.ts +++ b/packages/server/api/src/app/mcp/tools/ap-validate-step-config.ts @@ -1,7 +1,7 @@ import { BranchExecutionType, - McpServer, McpToolDefinition, + ProjectScopedMcpServer, RouterActionSettingsWithValidation, RouterExecutionType, SourceCode, @@ -10,7 +10,7 @@ import { FastifyBaseLogger } from 'fastify' import { z } from 'zod' import { mcpUtils } from './mcp-utils' -export const apValidateStepConfigTool = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition => { +export const apValidateStepConfigTool = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition => { return { title: 'ap_validate_step_config', description: 'Validate a step configuration before applying it. Returns field-level errors without modifying any flow. Use this to check your config is correct before calling ap_update_step or ap_update_trigger.', diff --git a/packages/server/api/src/app/mcp/tools/index.ts b/packages/server/api/src/app/mcp/tools/index.ts index b34b8a943a0..e692c8160a7 100644 --- a/packages/server/api/src/app/mcp/tools/index.ts +++ b/packages/server/api/src/app/mcp/tools/index.ts @@ -1,4 +1,4 @@ -import { McpServer, McpToolDefinition } from '@activepieces/shared' +import { McpToolDefinition, ProjectScopedMcpServer } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' import { apAddBranchTool } from './ap-add-branch' import { apAddStepTool } from './ap-add-step' @@ -84,7 +84,7 @@ export const ALL_CONTROLLABLE_TOOL_NAMES: string[] = [ 'ap_run_action', ] -export const activepiecesTools = (mcp: McpServer, log: FastifyBaseLogger): McpToolDefinition[] => [ +export const activepiecesTools = (mcp: ProjectScopedMcpServer, log: FastifyBaseLogger): McpToolDefinition[] => [ apBuildFlowTool(mcp, log), apCreateFlowTool(mcp, log), apDuplicateFlowTool(mcp, log), diff --git a/packages/server/api/src/app/server.ts b/packages/server/api/src/app/server.ts index f9c2361e8de..537c3c55f9a 100644 --- a/packages/server/api/src/app/server.ts +++ b/packages/server/api/src/app/server.ts @@ -18,7 +18,7 @@ import { exceptionHandler } from './helper/exception-handler' import { rejectedPromiseHandler } from './helper/promise-handler' import { system } from './helper/system/system' import { AppSystemProp } from './helper/system/system-props' -import { mcpOAuthHttpController } from './mcp/oauth/mcp-oauth.controller' +import { mcpOAuthHttpController, mcpPlatformHttpController } from './mcp/oauth/mcp-oauth.controller' import { mcpOAuthRootModule } from './mcp/oauth/mcp-oauth.module' @@ -31,6 +31,7 @@ export const setupServer = async (): Promise => { if (system.isApp()) { await app.register(mcpOAuthRootModule) await app.register(mcpOAuthHttpController, { prefix: '/mcp' }) + await app.register(mcpPlatformHttpController, { prefix: '/mcp/platform' }) } await app.register(async (apiApp) => { diff --git a/packages/server/api/src/app/workers/engine-controller.ts b/packages/server/api/src/app/workers/engine-controller.ts index 77a789c2391..6d20a8b2eae 100644 --- a/packages/server/api/src/app/workers/engine-controller.ts +++ b/packages/server/api/src/app/workers/engine-controller.ts @@ -2,8 +2,10 @@ import { FlowVersion, GetFlowVersionForWorkerRequest, ListFlowsRequest } from '@activepieces/shared' import { FastifyPluginAsyncZod } from 'fastify-type-provider-zod' import { StatusCodes } from 'http-status-codes' +import { z } from 'zod' import { entitiesMustBeOwnedByCurrentProject } from '../authentication/authorization' import { securityAccess } from '../core/security/authorization/fastify-security' +import { fileService } from '../file/file.service' import { flowService } from '../flows/flow/flow.service' import { flowVersionService } from '../flows/flow-version/flow-version.service' @@ -35,6 +37,17 @@ export const flowEngineWorker: FastifyPluginAsyncZod = async (app) => { return flowVersion }) + app.get('/files/:fileId', GetEnginePayloadFileRequest, async (request, reply) => { + const { data } = await fileService(request.log).getDataOrThrow({ + fileId: request.params.fileId, + projectId: request.principal.projectId, + }) + return reply + .type('application/octet-stream') + .status(StatusCodes.OK) + .send(data) + }) + } @@ -58,4 +71,15 @@ const GetLockedVersionRequest = { [StatusCodes.OK]: FlowVersion, }, }, +} + +const GetEnginePayloadFileRequest = { + config: { + security: securityAccess.engine(), + }, + schema: { + params: z.object({ + fileId: z.string(), + }), + }, } \ No newline at end of file diff --git a/packages/server/api/test/integration/cloud/mcp/mcp-rbac.test.ts b/packages/server/api/test/integration/cloud/mcp/mcp-rbac.test.ts index b8dbd7b87e1..ef300abffc5 100644 --- a/packages/server/api/test/integration/cloud/mcp/mcp-rbac.test.ts +++ b/packages/server/api/test/integration/cloud/mcp/mcp-rbac.test.ts @@ -3,16 +3,17 @@ import { FastifyBaseLogger, FastifyInstance } from 'fastify' import { apId, DefaultProjectRole, - McpServer, McpServerStatus, + McpServerType, Permission, + ProjectScopedMcpServer, } from '@activepieces/shared' import { setupTestEnvironment, teardownTestEnvironment } from '../../../helpers/test-setup' import { createMemberContext, createTestContext } from '../../../helpers/test-context' import { apCreateFlowTool } from '../../../../src/app/mcp/tools/ap-create-flow' import { apListFlowsTool } from '../../../../src/app/mcp/tools/ap-list-flows' import { apSetupGuideTool } from '../../../../src/app/mcp/tools/ap-setup-guide' -import { resolvePermissionChecker } from '../../../../src/app/mcp/mcp-service' +import { resolvePermissionChecker } from '../../../../src/app/mcp/mcp-permissions' let app: FastifyInstance let mockLog: FastifyBaseLogger @@ -26,12 +27,14 @@ afterAll(async () => { await teardownTestEnvironment() }) -function makeMcp(projectId: string): McpServer { +function makeMcp(projectId: string): ProjectScopedMcpServer { return { id: apId(), created: new Date().toISOString(), updated: new Date().toISOString(), projectId, + platformId: null, + type: McpServerType.PROJECT, status: McpServerStatus.ENABLED, token: apId(), enabledTools: null, diff --git a/packages/server/engine/src/lib/handler/context/engine-constants.ts b/packages/server/engine/src/lib/handler/context/engine-constants.ts index ab1b19ccd3b..936af415c4c 100644 --- a/packages/server/engine/src/lib/handler/context/engine-constants.ts +++ b/packages/server/engine/src/lib/handler/context/engine-constants.ts @@ -1,5 +1,5 @@ import { ContextVersion } from '@activepieces/pieces-framework' -import { DEFAULT_MCP_DATA, EngineGenericError, ExecuteFlowOperation, ExecutePropsOptions, ExecuteToolOperation, ExecuteTriggerOperation, ExecutionType, flowStructureUtil, FlowVersionState, PlatformId, Project, ProjectId, ResumePayload, RunEnvironment, StreamStepProgress, TriggerHookType } from '@activepieces/shared' +import { BeginExecuteFlowOperation, DEFAULT_MCP_DATA, EngineGenericError, ExecutePropsOptions, ExecuteToolOperation, ExecuteTriggerOperation, ExecutionState, ExecutionType, flowStructureUtil, FlowVersionState, PlatformId, Project, ProjectId, ResumeExecuteFlowOperation, ResumePayload, RunEnvironment, StreamStepProgress, TriggerHookType } from '@activepieces/shared' import { createPropsResolver, PropsResolver } from '../../variables/props-resolver' type RetryConstants = { @@ -118,7 +118,7 @@ export class EngineConstants { this.stepNames = params.stepNames } - public static fromExecuteFlowInput(input: ExecuteFlowOperation): EngineConstants { + public static fromExecuteFlowInput(input: ResolvedExecuteFlowOperation): EngineConstants { return new EngineConstants({ flowId: input.flowVersion.flowId, flowVersionId: input.flowVersion.id, @@ -136,7 +136,7 @@ export class EngineConstants { resumePayload: input.executionType === ExecutionType.RESUME ? input.resumePayload : undefined, runEnvironment: input.runEnvironment, stepNameToTest: input.stepNameToTest ?? undefined, - logsUploadUrl: input.logsUploadUrl, + logsUploadUrl: input.logsUploadUrl, logsFileId: input.logsFileId, timeoutInSeconds: input.timeoutInSeconds, platformId: input.platformId, @@ -250,4 +250,15 @@ export class EngineConstants { const addTrailingSlashIfMissing = (url: string): string => { return url.endsWith('/') ? url : url + '/' -} \ No newline at end of file +} + +export type ResolvedBeginExecuteFlowOperation = Omit & { + triggerPayload: unknown +} + +export type ResolvedResumeExecuteFlowOperation = Omit & { + resumePayload: ResumePayload + executionState: ExecutionState +} + +export type ResolvedExecuteFlowOperation = ResolvedBeginExecuteFlowOperation | ResolvedResumeExecuteFlowOperation diff --git a/packages/server/engine/src/lib/handler/flow-executor.ts b/packages/server/engine/src/lib/handler/flow-executor.ts index 7fac445f2e9..4cd3898131d 100644 --- a/packages/server/engine/src/lib/handler/flow-executor.ts +++ b/packages/server/engine/src/lib/handler/flow-executor.ts @@ -1,12 +1,12 @@ import { performance } from 'node:perf_hooks' -import { EngineGenericError, ExecuteFlowOperation, ExecutionType, FlowAction, FlowActionType, FlowRunStatus, FlowTrigger, GenericStepOutput, isNil, StepOutputStatus } from '@activepieces/shared' +import { EngineGenericError, ExecutionType, FlowAction, FlowActionType, FlowRunStatus, FlowTrigger, GenericStepOutput, isNil, StepOutputStatus } from '@activepieces/shared' import dayjs from 'dayjs' import { flowRunProgressReporter } from '../helper/flow-run-progress-reporter' import { loggingUtils } from '../helper/logging-utils' import { triggerHelper } from '../helper/trigger-helper' import { BaseExecutor } from './base-executor' import { codeExecutor } from './code-executor' -import { EngineConstants } from './context/engine-constants' +import { EngineConstants, ResolvedExecuteFlowOperation } from './context/engine-constants' import { FlowExecutorContext } from './context/flow-execution-context' import { loopExecutor } from './loop-executor' import { pieceExecutor } from './piece-executor' @@ -35,7 +35,7 @@ export const flowExecutor = { async executeFromTrigger({ executionState, constants, input }: { executionState: FlowExecutorContext constants: EngineConstants - input: ExecuteFlowOperation + input: ResolvedExecuteFlowOperation }): Promise { const trigger = input.flowVersion.trigger if (input.executionType === ExecutionType.BEGIN) { diff --git a/packages/server/engine/src/lib/helper/payload-file-client.ts b/packages/server/engine/src/lib/helper/payload-file-client.ts new file mode 100644 index 00000000000..15b895b6030 --- /dev/null +++ b/packages/server/engine/src/lib/helper/payload-file-client.ts @@ -0,0 +1,23 @@ +import { EngineGenericError } from '@activepieces/shared' + +export const payloadFileClient = { + get: async ({ apiUrl, engineToken, fileId }: GetPayloadFileRequest): Promise => { + const response = await fetch(`${apiUrl}v1/engine/files/${fileId}`, { + method: 'GET', + headers: { + Authorization: `Bearer ${engineToken}`, + }, + }) + if (!response.ok) { + throw new EngineGenericError('PayloadFileFetchError', `Failed to fetch payload file ${fileId}: ${response.status} ${response.statusText}`) + } + const arrayBuffer = await response.arrayBuffer() + return Buffer.from(arrayBuffer) + }, +} + +type GetPayloadFileRequest = { + apiUrl: string + engineToken: string + fileId: string +} diff --git a/packages/server/engine/src/lib/operations/flow.operation.ts b/packages/server/engine/src/lib/operations/flow.operation.ts index 09953f39e5f..f60a5f4c2ab 100644 --- a/packages/server/engine/src/lib/operations/flow.operation.ts +++ b/packages/server/engine/src/lib/operations/flow.operation.ts @@ -1,34 +1,37 @@ import { - BeginExecuteFlowOperation, EngineGenericError, EngineResponse, EngineResponseStatus, ExecuteFlowOperation, ExecuteTriggerResponse, + ExecutionState, ExecutionType, FlowActionType, FlowRunStatus, flowStructureUtil, GenericStepOutput, isNil, + JobPayload, LoopStepOutput, + ResumePayload, StepOutput, StepOutputStatus, TriggerHookType, TriggerPayload, } from '@activepieces/shared' -import { EngineConstants } from '../handler/context/engine-constants' +import { EngineConstants, ResolvedBeginExecuteFlowOperation, ResolvedExecuteFlowOperation } from '../handler/context/engine-constants' import { FlowExecutorContext } from '../handler/context/flow-execution-context' import { testExecutionContext } from '../handler/context/test-execution-context' import { flowExecutor } from '../handler/flow-executor' import { flowRunProgressReporter } from '../helper/flow-run-progress-reporter' +import { payloadFileClient } from '../helper/payload-file-client' import { triggerHelper } from '../helper/trigger-helper' export const flowOperation = { execute: async (operation: ExecuteFlowOperation): Promise> => { - const input = operation as ExecuteFlowOperation + const input = await resolveExecuteFlowOperation(operation) const constants = EngineConstants.fromExecuteFlowInput(input) - const output: FlowExecutorContext = (await executieSingleStepOrFlowOperation(input)).finishExecution() + const output: FlowExecutorContext = (await executieSingleStepOrFlowOperation(input, constants)).finishExecution() await flowRunProgressReporter.sendUpdate({ engineConstants: constants, flowExecutorContext: output, @@ -44,8 +47,7 @@ export const flowOperation = { }, } -const executieSingleStepOrFlowOperation = async (input: ExecuteFlowOperation): Promise => { - const constants = EngineConstants.fromExecuteFlowInput(input) +const executieSingleStepOrFlowOperation = async (input: ResolvedExecuteFlowOperation, constants: EngineConstants): Promise => { const testSingleStepMode = !isNil(constants.stepNameToTest) if (testSingleStepMode) { const testContext = await testExecutionContext.stateFromFlowVersion({ @@ -60,37 +62,27 @@ const executieSingleStepOrFlowOperation = async (input: ExecuteFlowOperation): P const step = flowStructureUtil.getActionOrThrow(input.stepNameToTest!, input.flowVersion.trigger) return flowExecutor.execute({ action: step, - executionState: await getFlowExecutionState(input, testContext), - constants: EngineConstants.fromExecuteFlowInput(input), + executionState: await getFlowExecutionState(input, constants, testContext), + constants, }) } return flowExecutor.executeFromTrigger({ - executionState: await getFlowExecutionState(input, FlowExecutorContext.empty()), + executionState: await getFlowExecutionState(input, constants, FlowExecutorContext.empty()), constants, input, }) } -async function getFlowExecutionState(input: ExecuteFlowOperation, flowContext: FlowExecutorContext): Promise { - switch (input.executionType) { - case ExecutionType.BEGIN: { - if (Object.keys(input.executionState.steps).length > 0) { - throw new EngineGenericError('InvalidBeginStateError', 'BEGIN operation received with non-empty execution state') - } - const newPayload = await runOrReturnPayload(input) - flowContext = flowContext.upsertStep(input.flowVersion.trigger.name, GenericStepOutput.create({ - type: input.flowVersion.trigger.type, - status: StepOutputStatus.SUCCEEDED, - input: {}, - }).setOutput(newPayload)) - break - } - case ExecutionType.RESUME: { - flowContext = flowContext.addTags(input.executionState.tags) - break - } +async function getFlowExecutionState(input: ResolvedExecuteFlowOperation, constants: EngineConstants, flowContext: FlowExecutorContext): Promise { + if (input.executionType === ExecutionType.BEGIN) { + const newPayload = await runOrReturnPayload(input, constants) + return flowContext.upsertStep(input.flowVersion.trigger.name, GenericStepOutput.create({ + type: input.flowVersion.trigger.type, + status: StepOutputStatus.SUCCEEDED, + input: {}, + }).setOutput(newPayload)) } - + flowContext = flowContext.addTags(input.executionState.tags) for (const [step, output] of Object.entries(input.executionState.steps)) { if ([StepOutputStatus.SUCCEEDED, StepOutputStatus.PAUSED].includes(output.status)) { const newOutput = await insertSuccessStepsOrPausedRecursively(output) @@ -102,7 +94,7 @@ async function getFlowExecutionState(input: ExecuteFlowOperation, flowContext: F return flowContext } -async function runOrReturnPayload(input: BeginExecuteFlowOperation): Promise { +async function runOrReturnPayload(input: ResolvedBeginExecuteFlowOperation, constants: EngineConstants): Promise { if (!input.executeTrigger) { return input.triggerPayload as TriggerPayload } @@ -114,7 +106,7 @@ async function runOrReturnPayload(input: BeginExecuteFlowOperation): Promise return newPayload.output[0] as TriggerPayload } @@ -141,4 +133,42 @@ async function insertSuccessStepsOrPausedRecursively(stepOutput: StepOutput): Pr return loopOutput.setIterations(newIterations) } return stepOutput -} \ No newline at end of file +} + +async function resolveExecuteFlowOperation(operation: ExecuteFlowOperation): Promise { + if (operation.executionType === ExecutionType.BEGIN) { + return { + ...operation, + triggerPayload: await resolveJobPayload(operation.triggerPayload, operation), + } + } + const executionState = await fetchExecutionStateFromLogs(operation.logsFileId, operation) + if (Object.keys(executionState.steps).length === 0) { + throw new EngineGenericError('EmptyResumeStateError', 'RESUME operation received with empty execution state') + } + return { + ...operation, + resumePayload: await resolveJobPayload(operation.resumePayload, operation) as ResumePayload, + executionState, + } +} + +async function resolveJobPayload(payload: JobPayload, operation: ExecuteFlowOperation): Promise { + if (payload.type === 'inline') { + return payload.value + } + const buffer = await payloadFileClient.get({ apiUrl: operation.internalApiUrl, engineToken: operation.engineToken, fileId: payload.fileId }) + return JSON.parse(buffer.toString('utf-8')) +} + +async function fetchExecutionStateFromLogs(logsFileId: string | undefined, operation: ExecuteFlowOperation): Promise { + if (isNil(logsFileId)) { + throw new EngineGenericError('ResumeLogsFileMissing', 'logsFileId is missing for RESUME operation') + } + const buffer = await payloadFileClient.get({ apiUrl: operation.internalApiUrl, engineToken: operation.engineToken, fileId: logsFileId }) + const parsed = JSON.parse(buffer.toString('utf-8')) + if (isNil(parsed?.executionState)) { + throw new EngineGenericError('ExecutionStateMissing', 'executionState is missing in logs file') + } + return parsed.executionState as ExecutionState +} diff --git a/packages/server/engine/test/handler/test-helper.ts b/packages/server/engine/test/handler/test-helper.ts index 42a74b5a6e4..e6d331efef1 100644 --- a/packages/server/engine/test/handler/test-helper.ts +++ b/packages/server/engine/test/handler/test-helper.ts @@ -1,5 +1,5 @@ import { ActionErrorHandlingOptions, BeginExecuteFlowOperation, BranchCondition, BranchExecutionType, CodeAction, ExecutionType, FlowAction, FlowActionType, FlowVersionState, LoopOnItemsAction, PieceAction, StreamStepProgress, PropertyExecutionType, RouterExecutionType, RunEnvironment } from '@activepieces/shared' -import { EngineConstants } from '../../src/lib/handler/context/engine-constants' +import { EngineConstants, ResolvedBeginExecuteFlowOperation } from '../../src/lib/handler/context/engine-constants' export const generateMockEngineConstants = (params?: Partial): EngineConstants => { return new EngineConstants( @@ -125,8 +125,8 @@ export function buildPieceAction({ name, input, skip, pieceName, actionName, nex } export function buildMockBeginExecuteFlowOperation( - params: Partial & Pick, -): BeginExecuteFlowOperation { + params: Partial & Pick, +): ResolvedBeginExecuteFlowOperation { return { projectId: 'projectId', engineToken: 'engineToken', @@ -137,7 +137,6 @@ export function buildMockBeginExecuteFlowOperation( flowRunId: 'flowRunId', executionType: ExecutionType.BEGIN, runEnvironment: RunEnvironment.TESTING, - executionState: { steps: {}, tags: [] }, workerHandlerId: null, httpRequestId: null, streamStepProgress: StreamStepProgress.NONE, diff --git a/packages/server/engine/test/operations/flow-operation-invariants.test.ts b/packages/server/engine/test/operations/flow-operation-invariants.test.ts index 4a15d346ccd..8e164a778d4 100644 --- a/packages/server/engine/test/operations/flow-operation-invariants.test.ts +++ b/packages/server/engine/test/operations/flow-operation-invariants.test.ts @@ -8,7 +8,7 @@ import { RunEnvironment, StepOutputStatus, } from '@activepieces/shared' -import type { BeginExecuteFlowOperation, FlowVersion } from '@activepieces/shared' +import type { BeginExecuteFlowOperation, FlowVersion, ResumeExecuteFlowOperation } from '@activepieces/shared' vi.mock('../../src/lib/helper/flow-run-progress-reporter', () => ({ flowRunProgressReporter: { @@ -17,6 +17,13 @@ vi.mock('../../src/lib/helper/flow-run-progress-reporter', () => ({ }, })) +const { mockGetPayloadFile } = vi.hoisted(() => ({ mockGetPayloadFile: vi.fn() })) +vi.mock('../../src/lib/helper/payload-file-client', () => ({ + payloadFileClient: { + get: mockGetPayloadFile, + }, +})) + import { flowOperation } from '../../src/lib/operations/flow.operation' function makeFlowVersion(): FlowVersion { @@ -56,51 +63,137 @@ function makeBeginOperation(overrides?: Partial): Beg flowRunId: 'run-1', executionType: ExecutionType.BEGIN, runEnvironment: RunEnvironment.TESTING, - executionState: { steps: {}, tags: [] }, workerHandlerId: null, httpRequestId: null, streamStepProgress: StreamStepProgress.NONE, stepNameToTest: null, - triggerPayload: {}, + triggerPayload: { type: 'inline', value: {} }, executeTrigger: false, ...overrides, } } +function makeResumeOperation(overrides?: Partial): ResumeExecuteFlowOperation { + return { + projectId: 'proj-1', + engineToken: 'test-token', + internalApiUrl: 'http://localhost:3000/', + publicApiUrl: 'http://localhost:4200/api/', + timeoutInSeconds: 600, + platformId: 'plat-1', + flowVersion: makeFlowVersion(), + flowRunId: 'run-1', + executionType: ExecutionType.RESUME, + runEnvironment: RunEnvironment.TESTING, + workerHandlerId: null, + httpRequestId: null, + streamStepProgress: StreamStepProgress.NONE, + stepNameToTest: null, + resumePayload: { type: 'inline', value: { data: {} } }, + logsFileId: 'logs-file-1', + ...overrides, + } +} + describe('flow operation invariants', () => { - describe('BEGIN execution state assertion', () => { - it('should throw EngineGenericError when BEGIN has non-empty execution state', async () => { - const operation = makeBeginOperation({ - executionState: { - steps: { - trigger_1: { - type: FlowTriggerType.EMPTY as any, - status: StepOutputStatus.SUCCEEDED, - input: {}, - output: {}, + describe('RESUME execution state hydration', () => { + it('throws EngineGenericError when RESUME has empty execution state in logs file', async () => { + mockGetPayloadFile.mockReset() + mockGetPayloadFile.mockResolvedValue( + Buffer.from(JSON.stringify({ executionState: { steps: {}, tags: [] } })), + ) + + const operation = makeResumeOperation() + + await expect(flowOperation.execute(operation)).rejects.toThrow(EngineGenericError) + await expect(flowOperation.execute(operation)).rejects.toThrow('RESUME operation received with empty execution state') + }) + + it('throws when logsFileId is missing on RESUME', async () => { + mockGetPayloadFile.mockReset() + const operation = makeResumeOperation({ logsFileId: undefined }) + + await expect(flowOperation.execute(operation)).rejects.toThrow(EngineGenericError) + await expect(flowOperation.execute(operation)).rejects.toThrow('logsFileId is missing for RESUME operation') + }) + + it('throws when executionState is missing in logs file', async () => { + mockGetPayloadFile.mockReset() + mockGetPayloadFile.mockResolvedValue(Buffer.from(JSON.stringify({}))) + + const operation = makeResumeOperation() + + await expect(flowOperation.execute(operation)).rejects.toThrow(EngineGenericError) + await expect(flowOperation.execute(operation)).rejects.toThrow('executionState is missing in logs file') + }) + + it('proceeds past hydration when logs file has non-empty execution state', async () => { + mockGetPayloadFile.mockReset() + mockGetPayloadFile.mockResolvedValue( + Buffer.from(JSON.stringify({ + executionState: { + steps: { + trigger_1: { + type: FlowTriggerType.EMPTY, + status: StepOutputStatus.SUCCEEDED, + input: {}, + output: {}, + }, }, + tags: [], }, - tags: [], - }, + })), + ) + + const operation = makeResumeOperation() + + try { + await flowOperation.execute(operation) + } + catch (e) { + expect((e as Error).message).not.toContain('empty execution state') + expect((e as Error).message).not.toContain('logsFileId is missing') + expect((e as Error).message).not.toContain('executionState is missing') + } + }) + }) + + describe('BEGIN payload hydration', () => { + it('inline payload is forwarded without calling getPayloadFile', async () => { + mockGetPayloadFile.mockReset() + const operation = makeBeginOperation({ + triggerPayload: { type: 'inline', value: { hello: 'world' } }, }) - await expect(flowOperation.execute(operation)).rejects.toThrow(EngineGenericError) - await expect(flowOperation.execute(operation)).rejects.toThrow('BEGIN operation received with non-empty execution state') + try { + await flowOperation.execute(operation) + } + catch { + // downstream may fail; we only assert RPC call shape + } + + expect(mockGetPayloadFile).not.toHaveBeenCalled() }) - it('should pass the assertion when BEGIN has empty execution state', async () => { + it('ref payload is fetched via the engine HTTP client', async () => { + mockGetPayloadFile.mockReset() + mockGetPayloadFile.mockResolvedValue(Buffer.from(JSON.stringify({ hello: 'ref' }))) const operation = makeBeginOperation({ - executionState: { steps: {}, tags: [] }, + triggerPayload: { type: 'ref', fileId: 'payload-file-1' }, }) - // The operation will fail further downstream (trigger setup), - // but it should NOT throw InvalidBeginStateError try { await flowOperation.execute(operation) } - catch (e) { - expect((e as Error).name).not.toBe('InvalidBeginStateError') + catch { + // downstream may fail; we only assert RPC call shape } + + expect(mockGetPayloadFile).toHaveBeenCalledWith({ + apiUrl: 'http://localhost:3000/', + engineToken: 'test-token', + fileId: 'payload-file-1', + }) }) }) }) diff --git a/packages/server/worker/src/lib/execute/jobs/execute-flow.ts b/packages/server/worker/src/lib/execute/jobs/execute-flow.ts index 5f4a13770de..c5920631cd6 100644 --- a/packages/server/worker/src/lib/execute/jobs/execute-flow.ts +++ b/packages/server/worker/src/lib/execute/jobs/execute-flow.ts @@ -7,23 +7,19 @@ import { EngineResponseStatus, ErrorCode, ExecuteFlowJobData, - ExecutionState, ExecutionType, FlowRunStatus, FlowVersion, isNil, ResumeExecuteFlowOperation, - ResumePayload, tryCatch, WorkerJobType, - WorkerToApiContract, } from '@activepieces/shared' import { flowCache } from '../../cache/flow/flow-cache' import { system, WorkerSystemProp } from '../../config/configs' import { workerSettings } from '../../config/worker-settings' import { FireAndForgetJobResult, JobContext, JobHandler, JobResultKind } from '../types' import { provisionFlowPieces } from '../utils/flow-helpers' -import { resolvePayload } from '../utils/resolve-payload' export const executeFlowJob: JobHandler = { jobType: WorkerJobType.EXECUTE_FLOW, @@ -47,6 +43,14 @@ export const executeFlowJob: JobHandler { +): BeginExecuteFlowOperation | ResumeExecuteFlowOperation { const base = { flowVersion, flowRunId: data.runId, @@ -126,52 +128,22 @@ async function buildFlowOperation( } if (data.executionType === ExecutionType.RESUME) { - const executionState = await fetchExecutionState({ apiClient: ctx.apiClient, data }) - if (Object.keys(executionState.steps).length === 0) { - ctx.log.error({ runId: data.runId, executionType: data.executionType }, 'RESUME operation has empty execution state — this is a bug that would cause an infinite loop') - throw new ActivepiecesError({ - code: ErrorCode.VALIDATION, - params: { - message: 'RESUME operation received with empty execution state', - }, - }) - } return { ...base, executionType: ExecutionType.RESUME, - executionState, - resumePayload: resolvedPayload as ResumePayload, + resumePayload: data.payload, } } return { ...base, executionType: ExecutionType.BEGIN, - executionState: { steps: {}, tags: [] }, - triggerPayload: resolvedPayload, + triggerPayload: data.payload, executeTrigger: data.executeTrigger ?? false, sampleData: data.sampleData, } } -async function fetchExecutionState({ apiClient, data }: { apiClient: WorkerToApiContract, data: ExecuteFlowJobData }): Promise { - if (isNil(data.logsFileId)) { - throw new ActivepiecesError({ - code: ErrorCode.RESUME_LOGS_FILE_MISSING, - params: { runId: data.runId }, - }, 'logsFileId is missing for RESUME operation') - } - const buffer = await apiClient.getPayloadFile({ fileId: data.logsFileId, projectId: data.projectId }) - const parsed = JSON.parse(buffer.toString('utf-8')) - if (isNil(parsed.executionState)) { - throw new ActivepiecesError({ - code: ErrorCode.EXECUTION_STATE_MISSING, - params: { logsFileId: data.logsFileId }, - }, 'executionState is missing in logs file') - } - return parsed.executionState -} - async function reportFlowStatus( ctx: JobContext, data: ExecuteFlowJobData, diff --git a/packages/server/worker/test/lib/execute/jobs/execute-flow.test.ts b/packages/server/worker/test/lib/execute/jobs/execute-flow.test.ts index 09a8aa19cd3..6bcf5a6c4ea 100644 --- a/packages/server/worker/test/lib/execute/jobs/execute-flow.test.ts +++ b/packages/server/worker/test/lib/execute/jobs/execute-flow.test.ts @@ -9,7 +9,6 @@ import { FlowVersionState, StreamStepProgress, RunEnvironment, - StepOutputStatus, WorkerJobType, } from '@activepieces/shared' import type { ExecuteFlowJobData, FlowVersion } from '@activepieces/shared' @@ -32,10 +31,6 @@ vi.mock('../../../../src/lib/execute/utils/flow-helpers', () => ({ provisionFlowPieces: vi.fn().mockResolvedValue(true), })) -vi.mock('../../../../src/lib/execute/utils/resolve-payload', () => ({ - resolvePayload: vi.fn().mockImplementation((payload: unknown) => Promise.resolve(payload)), -})) - import { executeFlowJob } from '../../../../src/lib/execute/jobs/execute-flow' import { JobResultKind } from '../../../../src/lib/execute/types' @@ -95,7 +90,7 @@ function makeResumeJobData(overrides?: Partial): ExecuteFlow flowId: 'flow-1', flowVersionId: 'fv-1', runId: 'run-1', - payload: {}, + payload: { type: 'inline', value: {} }, executionType: ExecutionType.RESUME, streamStepProgress: StreamStepProgress.NONE, logsUploadUrl: 'http://example.com/upload', @@ -107,7 +102,7 @@ function makeResumeJobData(overrides?: Partial): ExecuteFlow function makeMockContext(apiOverrides?: Record) { const mockSandbox = { start: vi.fn(), - execute: vi.fn().mockResolvedValue({ engine: { status: 'OK' } }), + execute: vi.fn().mockResolvedValue({ status: 'OK' }), } return { log: { @@ -129,6 +124,7 @@ function makeMockContext(apiOverrides?: Record) { engineToken: 'test-token', internalApiUrl: 'http://localhost:3000', publicApiUrl: 'http://localhost:4200', + mockSandbox, } as any } @@ -137,14 +133,56 @@ describe('executeFlowJob', () => { mockGetVersion.mockResolvedValue(makeFlowVersion()) }) - describe('RESUME execution state validation', () => { - it('should throw VALIDATION error when RESUME has empty execution state', async () => { + describe('payload pass-through (no worker-side fetch)', () => { + it('does not call getPayloadFile in the worker — payload resolution is deferred to the engine', async () => { const ctx = makeMockContext() - ctx.apiClient.getPayloadFile.mockResolvedValue( - Buffer.from(JSON.stringify({ executionState: { steps: {}, tags: [] } })), - ) + const data = makeResumeJobData({ + executionType: ExecutionType.BEGIN, + payload: { type: 'ref', fileId: 'huge-file-1' }, + }) + + await executeFlowJob.execute(ctx, data) + + expect(ctx.apiClient.getPayloadFile).not.toHaveBeenCalled() + }) + + it('forwards the JobPayload ref unchanged to the engine for BEGIN', async () => { + const ctx = makeMockContext() + const data = makeResumeJobData({ + executionType: ExecutionType.BEGIN, + payload: { type: 'ref', fileId: 'huge-file-1' }, + }) + + await executeFlowJob.execute(ctx, data) + + const operation = ctx.mockSandbox.execute.mock.calls[0][1] + expect(operation.executionType).toBe(ExecutionType.BEGIN) + expect(operation.triggerPayload).toEqual({ type: 'ref', fileId: 'huge-file-1' }) + expect(operation.executionState).toBeUndefined() + }) - const data = makeResumeJobData() + it('forwards the JobPayload ref unchanged to the engine for RESUME and never reads logsFileId', async () => { + const ctx = makeMockContext() + const data = makeResumeJobData({ + payload: { type: 'ref', fileId: 'resume-payload-1' }, + logsFileId: 'logs-file-1', + }) + + await executeFlowJob.execute(ctx, data) + + const operation = ctx.mockSandbox.execute.mock.calls[0][1] + expect(operation.executionType).toBe(ExecutionType.RESUME) + expect(operation.resumePayload).toEqual({ type: 'ref', fileId: 'resume-payload-1' }) + expect(operation.logsFileId).toBe('logs-file-1') + expect(operation.executionState).toBeUndefined() + expect(ctx.apiClient.getPayloadFile).not.toHaveBeenCalled() + }) + }) + + describe('RESUME validation', () => { + it('still throws when logsFileId is missing for RESUME', async () => { + const ctx = makeMockContext() + const data = makeResumeJobData({ logsFileId: undefined as unknown as string }) try { await executeFlowJob.execute(ctx, data) @@ -152,49 +190,17 @@ describe('executeFlowJob', () => { } catch (e) { expect(e).toBeInstanceOf(ActivepiecesError) - expect((e as ActivepiecesError).error.code).toBe(ErrorCode.VALIDATION) + expect((e as ActivepiecesError).error.code).toBe(ErrorCode.RESUME_LOGS_FILE_MISSING) } - expect(ctx.log.error).toHaveBeenCalledWith( - expect.objectContaining({ runId: 'run-1' }), - expect.stringContaining('empty execution state'), - ) - expect(ctx.apiClient.uploadRunLog).toHaveBeenCalledWith( expect.objectContaining({ status: FlowRunStatus.INTERNAL_ERROR }), ) }) - - it('should proceed normally when RESUME has non-empty execution state', async () => { - const ctx = makeMockContext() - ctx.apiClient.getPayloadFile.mockResolvedValue( - Buffer.from(JSON.stringify({ - executionState: { - steps: { - trigger_1: { - type: FlowTriggerType.EMPTY, - status: StepOutputStatus.SUCCEEDED, - input: {}, - output: {}, - }, - }, - tags: [], - }, - })), - ) - - const data = makeResumeJobData() - const result = await executeFlowJob.execute(ctx, data) - - expect(result).toBeDefined() - expect(ctx.apiClient.uploadRunLog).not.toHaveBeenCalledWith( - expect.objectContaining({ status: FlowRunStatus.INTERNAL_ERROR }), - ) - }) }) describe('missing piece handling', () => { - it('should mark run as FAILED and skip sandbox when flow version is not found (missing piece)', async () => { + it('marks run as FAILED and skips sandbox when flow version is not found', async () => { mockGetVersion.mockResolvedValue(null) const ctx = makeMockContext() diff --git a/packages/shared/package.json b/packages/shared/package.json index 3798a779700..735b2320da9 100644 --- a/packages/shared/package.json +++ b/packages/shared/package.json @@ -1,6 +1,6 @@ { "name": "@activepieces/shared", - "version": "0.70.1", + "version": "0.71.1", "type": "commonjs", "sideEffects": false, "main": "./dist/src/index.js", diff --git a/packages/shared/src/lib/automation/engine/engine-operation.ts b/packages/shared/src/lib/automation/engine/engine-operation.ts index d7e8dae2852..db92861d7e8 100644 --- a/packages/shared/src/lib/automation/engine/engine-operation.ts +++ b/packages/shared/src/lib/automation/engine/engine-operation.ts @@ -3,11 +3,12 @@ import { PlatformId } from '../../management/platform' import { ProjectId } from '../../management/project/project' import { ExecutionToolStatus, PredefinedInputsStructure } from '../agents' import { AppConnectionValue } from '../app-connection/app-connection' -import { ExecutionState, ExecutionType, ResumePayload } from '../flow-run/execution/execution-output' +import { ExecutionType } from '../flow-run/execution/execution-output' import { FlowRunId, RunEnvironment } from '../flow-run/flow-run' import { FlowVersion } from '../flows/flow-version' import { PiecePackage } from '../pieces' import { ScheduleOptions } from '../trigger' +import { JobPayload } from '../workers/job-data' export enum EngineOperationType { EXTRACT_PIECE_METADATA = 'EXTRACT_PIECE_METADATA', @@ -89,7 +90,6 @@ type BaseExecuteFlowOperation = BaseEngineOperation & { flowRunId: FlowRunId executionType: T runEnvironment: RunEnvironment - executionState: ExecutionState workerHandlerId: string | null httpRequestId: string | null streamStepProgress: StreamStepProgress @@ -105,12 +105,12 @@ export enum StreamStepProgress { } export type BeginExecuteFlowOperation = BaseExecuteFlowOperation & { - triggerPayload: unknown + triggerPayload: JobPayload executeTrigger: boolean } export type ResumeExecuteFlowOperation = BaseExecuteFlowOperation & { - resumePayload: ResumePayload + resumePayload: JobPayload } export type ExecuteFlowOperation = BeginExecuteFlowOperation | ResumeExecuteFlowOperation diff --git a/packages/shared/src/lib/automation/mcp/mcp-oauth.ts b/packages/shared/src/lib/automation/mcp/mcp-oauth.ts index 6123c6b1935..9c7b6e52075 100644 --- a/packages/shared/src/lib/automation/mcp/mcp-oauth.ts +++ b/packages/shared/src/lib/automation/mcp/mcp-oauth.ts @@ -20,7 +20,7 @@ export const McpOAuthToken = z.object({ refreshToken: z.string(), clientId: z.string(), userId: z.string(), - projectId: z.string(), + projectId: z.string().nullable(), platformId: z.string(), scopes: z.array(z.string()).nullable(), expiresAt: z.string(), @@ -34,7 +34,7 @@ export const McpOAuthAuthorizationCode = z.object({ code: z.string(), clientId: z.string(), userId: z.string(), - projectId: z.string(), + projectId: z.string().nullable(), platformId: z.string(), redirectUri: z.string(), codeChallenge: z.string(), diff --git a/packages/shared/src/lib/automation/mcp/mcp.ts b/packages/shared/src/lib/automation/mcp/mcp.ts index accd0999d1b..5b0d09ca21c 100644 --- a/packages/shared/src/lib/automation/mcp/mcp.ts +++ b/packages/shared/src/lib/automation/mcp/mcp.ts @@ -13,9 +13,16 @@ export enum McpServerStatus { DISABLED = 'DISABLED', } +export enum McpServerType { + PLATFORM = 'PLATFORM', + PROJECT = 'PROJECT', +} + export const McpServer = z.object({ ...BaseModelSchema, - projectId: ApId, + platformId: ApId.nullable(), + projectId: ApId.nullable(), + type: z.enum([McpServerType.PLATFORM, McpServerType.PROJECT]), status: z.nativeEnum(McpServerStatus), token: ApId, enabledTools: z.array(z.string()).nullable(), @@ -28,6 +35,7 @@ export type PopulatedMcpServer = z.infer export type McpServer = z.infer +export type ProjectScopedMcpServer = McpServer & { projectId: string } export const UpdateMcpServerRequest = z.object({ status: z.nativeEnum(McpServerStatus).optional(), diff --git a/packages/shared/src/lib/core/common/telemetry.ts b/packages/shared/src/lib/core/common/telemetry.ts index 7aed83cf1ca..4f8d7c28871 100644 --- a/packages/shared/src/lib/core/common/telemetry.ts +++ b/packages/shared/src/lib/core/common/telemetry.ts @@ -142,8 +142,9 @@ type McpToolCalled = { } type McpServerConnected = { - projectId: string userId: string + projectId?: string + platformId?: string } type PieceSelectorSearch = { diff --git a/packages/web/public/locales/en/translation.json b/packages/web/public/locales/en/translation.json index 266bda5948e..d996ccb881f 100644 --- a/packages/web/public/locales/en/translation.json +++ b/packages/web/public/locales/en/translation.json @@ -1497,6 +1497,7 @@ "Upgrade anyway": "Upgrade anyway", "Authorize Application": "Authorize Application", "wants to access your Activepieces account": "wants to access your Activepieces account", + "wants to connect to your Activepieces account": "wants to connect to your Activepieces account", "This will allow the app to execute automations and access tools in your selected project.": "This will allow the app to execute automations and access tools in your selected project.", "Select Project": "Select Project", "Loading projects...": "Loading projects...", @@ -1505,6 +1506,12 @@ "Deny": "Deny", "Authorize": "Authorize", "Unknown app": "Unknown app", + "Build, test, and manage automations": "Build, test, and manage automations", + "Use connections and execute flows": "Use connections and execute flows", + "Connected": "Connected", + "is now connected to your project.": "is now connected to your project.", + "is now connected to your platform.": "is now connected to your platform.", + "You can close this tab and return to the application.": "You can close this tab and return to the application.", "Embedding model for knowledge base: {model}": "Embedding model for knowledge base: {model}", "This provider does not support knowledge base embeddings.": "This provider does not support knowledge base embeddings.", "Knowledge base requires a provider that supports embeddings, such as OpenAI or Google.": "Knowledge base requires a provider that supports embeddings, such as OpenAI or Google.", diff --git a/packages/web/src/app/components/project-settings/mcp-server/index.tsx b/packages/web/src/app/components/project-settings/mcp-server/index.tsx index 88b55f8254a..1bf21ca25d8 100644 --- a/packages/web/src/app/components/project-settings/mcp-server/index.tsx +++ b/packages/web/src/app/components/project-settings/mcp-server/index.tsx @@ -86,7 +86,13 @@ export const McpServerSettings = () => { 'Control which built-in Activepieces tools are available to agents via this MCP server.', )}

- + + updateMcpServer({ enabledTools: tools }) + } + />
diff --git a/packages/web/src/app/components/project-settings/mcp-server/mcp-tools.tsx b/packages/web/src/app/components/project-settings/mcp-server/mcp-tools.tsx index 6bcad29bf18..17cc8ba1eaf 100644 --- a/packages/web/src/app/components/project-settings/mcp-server/mcp-tools.tsx +++ b/packages/web/src/app/components/project-settings/mcp-server/mcp-tools.tsx @@ -1,4 +1,3 @@ -import { PopulatedMcpServer } from '@activepieces/shared'; import { t } from 'i18next'; import { Lock } from 'lucide-react'; import { useEffect, useState } from 'react'; @@ -16,37 +15,36 @@ import { TooltipContent, TooltipTrigger, } from '@/components/ui/tooltip'; -import { authenticationSession } from '@/lib/authentication-session'; import { cn } from '@/lib/utils'; -import { mcpHooks } from './utils/mcp-hooks'; import { ALL_CONTROLLABLE_TOOL_NAMES, TOOL_CATEGORIES, } from './utils/mcp-tools-metadata'; type McpToolsProps = { - mcpServer: PopulatedMcpServer; + enabledTools: string[] | null; + isPending: boolean; + onUpdateEnabledTools: (tools: string[]) => void; }; -export function McpTools({ mcpServer }: McpToolsProps) { - const currentProjectId = authenticationSession.getProjectId(); - const { mutate: updateMcpServer, isPending } = mcpHooks.useUpdateMcpServer( - currentProjectId!, - ); - +export function McpTools({ + enabledTools: externalEnabledTools, + isPending, + onUpdateEnabledTools, +}: McpToolsProps) { const [enabledTools, setEnabledTools] = useState( - () => mcpServer.enabledTools ?? ALL_CONTROLLABLE_TOOL_NAMES, + () => externalEnabledTools ?? ALL_CONTROLLABLE_TOOL_NAMES, ); useEffect(() => { if (!isPending) { - setEnabledTools(mcpServer.enabledTools ?? ALL_CONTROLLABLE_TOOL_NAMES); + setEnabledTools(externalEnabledTools ?? ALL_CONTROLLABLE_TOOL_NAMES); } - }, [mcpServer.enabledTools, isPending]); + }, [externalEnabledTools, isPending]); const saveEnabledTools = useDebouncedCallback((tools: string[]) => { - updateMcpServer({ enabledTools: tools }); + onUpdateEnabledTools(tools); }, 300); const toggleTool = (name: string, checked: boolean) => { diff --git a/packages/web/src/app/components/sidebar/platform/index.tsx b/packages/web/src/app/components/sidebar/platform/index.tsx index 8514a622263..92e44ea3749 100644 --- a/packages/web/src/app/components/sidebar/platform/index.tsx +++ b/packages/web/src/app/components/sidebar/platform/index.tsx @@ -58,6 +58,11 @@ export function PlatformSidebar() { label: t('AI Providers'), icon: BotIcon, }, + { + to: '/platform/setup/mcp', + label: t('MCP Server'), + icon: ServerIcon, + }, { to: '/platform/setup/branding', label: t('Branding'), diff --git a/packages/web/src/app/routes/mcp-authorize/index.tsx b/packages/web/src/app/routes/mcp-authorize/index.tsx index 9eb309f2cf2..c59c319ee1f 100644 --- a/packages/web/src/app/routes/mcp-authorize/index.tsx +++ b/packages/web/src/app/routes/mcp-authorize/index.tsx @@ -26,7 +26,7 @@ import { authenticationSession } from '@/lib/authentication-session'; function McpAuthorizePage() { const [searchParams] = useSearchParams(); const authRequestId = searchParams.get('authRequestId'); - const clientName = decodeJwtClientName(authRequestId); + const { clientName, isPlatformScoped } = decodeJwtPayload(authRequestId); const [selectedProjectId, setSelectedProjectId] = useState< string | undefined >(undefined); @@ -48,11 +48,11 @@ function McpAuthorizePage() { ...(searchValue && { displayName: searchValue }), ...(selectedTypes.length > 0 && { types: selectedTypes }), }), - enabled: isLoggedIn && !!authRequestId, + enabled: isLoggedIn && !!authRequestId && !isPlatformScoped, }); const approveMutation = useMutation({ - mutationFn: (body: { authRequestId: string; projectId: string }) => + mutationFn: (body: { authRequestId: string; projectId?: string }) => api.post<{ redirectUrl: string }>('/v1/mcp-oauth/approve', body), onSuccess: (data) => { window.location.href = data.redirectUrl; @@ -79,10 +79,10 @@ function McpAuthorizePage() { } const handleAuthorize = () => { - if (!selectedProjectId) return; + if (!isPlatformScoped && !selectedProjectId) return; approveMutation.mutate({ authRequestId, - projectId: selectedProjectId, + ...(selectedProjectId && { projectId: selectedProjectId }), }); }; @@ -101,7 +101,9 @@ function McpAuthorizePage() { {clientName} {' '} - {t('is now connected to your project.')} + {isPlatformScoped + ? t('is now connected to your platform.') + : t('is now connected to your project.')}
@@ -145,43 +147,45 @@ function McpAuthorizePage() { -
-
- - } - options={projectTypeOptions} - selectedValues={selectedTypes} - onChange={setSelectedTypes} + {!isPlatformScoped && ( +
+
+ + } + options={projectTypeOptions} + selectedValues={selectedTypes} + onChange={setSelectedTypes} + /> +
+ + options={options} + onChange={(value) => setSelectedProjectId(value ?? undefined)} + value={selectedProjectId} + placeholder={t('Search projects...')} + disabled={projectsLoading} + loading={projectsLoading} + refreshOnSearch={debouncedSetSearchValue} + valuesRendering={(value) => { + const project = projectsMap.get(String(value)); + if (!project) return null; + return ( +
+ {project.displayName} + + {project.type === ProjectType.PERSONAL + ? t('Personal') + : t('Team')} + +
+ ); + }} />
- - options={options} - onChange={(value) => setSelectedProjectId(value ?? undefined)} - value={selectedProjectId} - placeholder={t('Search projects...')} - disabled={projectsLoading} - loading={projectsLoading} - refreshOnSearch={debouncedSetSearchValue} - valuesRendering={(value) => { - const project = projectsMap.get(String(value)); - if (!project) return null; - return ( -
- {project.displayName} - - {project.type === ProjectType.PERSONAL - ? t('Personal') - : t('Team')} - -
- ); - }} - /> -
+ )} {approveMutation.isError && (
@@ -202,7 +206,7 @@ function McpAuthorizePage() { type="button" className="flex-1" loading={approveMutation.isPending} - disabled={!selectedProjectId} + disabled={!isPlatformScoped && !selectedProjectId} onClick={handleAuthorize} > {t('Authorize')} @@ -231,13 +235,22 @@ function PermissionItem({ ); } -function decodeJwtClientName(token: string | null): string { +function decodeJwtPayload(token: string | null): { + clientName: string; + isPlatformScoped: boolean; +} { try { - if (!token) return t('Unknown app'); - const payload = jwtDecode<{ clientName?: string }>(token); - return payload.clientName ?? t('Unknown app'); + if (!token) + return { clientName: t('Unknown app'), isPlatformScoped: false }; + const payload = jwtDecode<{ clientName?: string; resource?: string }>( + token, + ); + const clientName = payload.clientName ?? t('Unknown app'); + const isPlatformScoped = + payload.resource?.endsWith('/mcp/platform') ?? false; + return { clientName, isPlatformScoped }; } catch { - return t('Unknown app'); + return { clientName: t('Unknown app'), isPlatformScoped: false }; } } diff --git a/packages/web/src/app/routes/platform-routes.tsx b/packages/web/src/app/routes/platform-routes.tsx index 1ee9d9a520c..c4492974143 100644 --- a/packages/web/src/app/routes/platform-routes.tsx +++ b/packages/web/src/app/routes/platform-routes.tsx @@ -42,6 +42,7 @@ const SSOPage = React.lazy(() => import('./platform/security/sso').then((m) => ({ default: m.SSOPage })), ); const AIProvidersPage = React.lazy(() => import('./platform/setup/ai')); +const PlatformMcpPage = React.lazy(() => import('./platform/setup/mcp')); const BrandingPage = React.lazy(() => import('./platform/setup/branding').then((m) => ({ default: m.BrandingPage, @@ -125,6 +126,18 @@ export const platformRoutes = [ ), }, + { + path: '/platform/setup/mcp', + element: ( + + + + + + + + ), + }, { path: '/platform/setup/pieces', element: ( diff --git a/packages/web/src/app/routes/platform/setup/mcp/index.tsx b/packages/web/src/app/routes/platform/setup/mcp/index.tsx new file mode 100644 index 00000000000..730b1cd1ac3 --- /dev/null +++ b/packages/web/src/app/routes/platform/setup/mcp/index.tsx @@ -0,0 +1,152 @@ +import { ApFlagId, McpServerStatus } from '@activepieces/shared'; +import { t } from 'i18next'; + +import { CenteredPage } from '@/app/components/centered-page'; +import { McpTools } from '@/app/components/project-settings/mcp-server/mcp-tools'; +import { CopyToClipboardInput } from '@/components/custom/clipboard/copy-to-clipboard'; +import { CollapsibleJson } from '@/components/custom/collapsible-json'; +import { + Field, + FieldContent, + FieldDescription, + FieldLabel, +} from '@/components/custom/field'; +import { LoadingSpinner } from '@/components/custom/spinner'; +import { Switch } from '@/components/ui/switch'; +import { Tabs, TabsContent, TabsList, TabsTrigger } from '@/components/ui/tabs'; +import { flagsHooks } from '@/hooks/flags-hooks'; + +import { platformMcpHooks } from './platform-mcp-hooks'; + +export default function PlatformMcpPage() { + const { data: mcpServer, isLoading } = + platformMcpHooks.usePlatformMcpServer(); + const { mutate: updateStatus, isPending: isStatusUpdating } = + platformMcpHooks.useUpdatePlatformMcpStatus(); + const { mutate: updateTools, isPending: isToolsUpdating } = + platformMcpHooks.useUpdatePlatformMcpTools(); + const { data: publicUrl } = flagsHooks.useFlag(ApFlagId.PUBLIC_URL); + + if (isLoading) { + return ( + +
+ +
+ + ); + } + + const isEnabled = mcpServer?.status === McpServerStatus.ENABLED; + const serverUrl = `${(publicUrl ?? '').replace(/\/$/, '')}/mcp/platform`; + + const jsonConfiguration = { + mcpServers: { + activepieces: { + url: serverUrl, + }, + }, + }; + + return ( + +
+ + + + {t('Enable Platform MCP')} + + + {t( + 'Allow the AI Chat and external agents to access tools across all projects on this platform.', + )} + + + + updateStatus({ + status: checked + ? McpServerStatus.ENABLED + : McpServerStatus.DISABLED, + }) + } + disabled={isStatusUpdating} + /> + + + {isEnabled && mcpServer && ( + + + {t('Connection')} + {t('Tools')} + + + +
+
+ +

+ {t( + 'Use this URL to connect from Cursor, Windsurf, Claude Desktop, or any MCP-compatible client. Authentication is handled via OAuth.', + )} +

+ +
+ + +
+ + + +
+

+ {t('Internal Tools')} +

+

+ {t( + 'Control which built-in tools are available to the AI Chat and external agents via the platform MCP server.', + )} +

+ + updateTools({ enabledTools: tools }) + } + /> +
+ + + )} +
+ + ); +} diff --git a/packages/web/src/app/routes/platform/setup/mcp/platform-mcp-api.ts b/packages/web/src/app/routes/platform/setup/mcp/platform-mcp-api.ts new file mode 100644 index 00000000000..ac3e4a33126 --- /dev/null +++ b/packages/web/src/app/routes/platform/setup/mcp/platform-mcp-api.ts @@ -0,0 +1,16 @@ +import { McpServer, UpdateMcpServerRequest } from '@activepieces/shared'; + +import { api } from '@/lib/api'; + +async function get(): Promise { + return api.get('/v1/mcp-server'); +} + +async function update(request: UpdateMcpServerRequest): Promise { + return api.post('/v1/mcp-server', request); +} + +export const platformMcpApi = { + get, + update, +}; diff --git a/packages/web/src/app/routes/platform/setup/mcp/platform-mcp-hooks.ts b/packages/web/src/app/routes/platform/setup/mcp/platform-mcp-hooks.ts new file mode 100644 index 00000000000..73c3f1404cc --- /dev/null +++ b/packages/web/src/app/routes/platform/setup/mcp/platform-mcp-hooks.ts @@ -0,0 +1,36 @@ +import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query'; + +import { platformMcpApi } from './platform-mcp-api'; + +const QUERY_KEY = ['platform-mcp-server']; + +export const platformMcpHooks = { + usePlatformMcpServer() { + return useQuery({ + queryKey: QUERY_KEY, + queryFn: () => platformMcpApi.get(), + retry: false, + meta: { showErrorDialog: true, loadSubsetOptions: {} }, + }); + }, + + useUpdatePlatformMcpStatus() { + const queryClient = useQueryClient(); + return useMutation({ + mutationFn: platformMcpApi.update, + onSuccess: (data) => { + queryClient.setQueryData(QUERY_KEY, data); + }, + }); + }, + + useUpdatePlatformMcpTools() { + const queryClient = useQueryClient(); + return useMutation({ + mutationFn: platformMcpApi.update, + onSuccess: (data) => { + queryClient.setQueryData(QUERY_KEY, data); + }, + }); + }, +}; diff --git a/packages/web/vite.config.mts b/packages/web/vite.config.mts index dbe4a9f8138..7ae490d353f 100644 --- a/packages/web/vite.config.mts +++ b/packages/web/vite.config.mts @@ -29,7 +29,7 @@ export default defineConfig(({ command, mode }) => { }, ws: true, }, - '^/mcp$': { + '^/mcp(/|$)': { target: 'http://127.0.0.1:3000', secure: false, changeOrigin: true,