wip: skyfire pay id token integation, add progress notif integration test

Author: MQ37

src/actor/server.ts

@@ -12,6 +12,7 @@ import express from 'express';
 import log from '@apify/log';
 
 import { ActorsMcpServer } from '../mcp/server.js';
+import type { AuthToken } from '../types.js';
 import { getHelpMessage, HEADER_READINESS_PROBE, Routes, TransportType } from './const.js';
 import { getActorRunData } from './utils.js';
 
@@ -73,9 +74,12 @@ export function createExpressApp(
             const transport = new SSEServerTransport(Routes.MESSAGE, res);
 
             // Load MCP server tools
-            const apifyToken = process.env.APIFY_TOKEN as string;
+            const authToken: AuthToken = {
+                value: process.env.APIFY_TOKEN as string,
+                type: 'apify',
+            };
             log.debug('Loading tools from URL', { sessionId: transport.sessionId, tr: TransportType.SSE });
-            await mcpServer.loadToolsFromUrl(req.url, apifyToken);
+            await mcpServer.loadToolsFromUrl(req.url, authToken);
 
             transportsSSE[transport.sessionId] = transport;
             mcpServers[transport.sessionId] = mcpServer;
@@ -155,9 +159,12 @@ export function createExpressApp(
                 const mcpServer = new ActorsMcpServer(false);
 
                 // Load MCP server tools
-                const apifyToken = process.env.APIFY_TOKEN as string;
+                const authToken: AuthToken = {
+                    value: process.env.APIFY_TOKEN as string,
+                    type: 'apify',
+                };
                 log.debug('Loading tools from URL', { sessionId: transport.sessionId, tr: TransportType.HTTP });
-                await mcpServer.loadToolsFromUrl(req.url, apifyToken);
+                await mcpServer.loadToolsFromUrl(req.url, authToken);
 
                 // Connect the transport to the MCP server BEFORE handling the request
                 await mcpServer.connect(transport);

src/apify-client.ts

@@ -3,6 +3,7 @@ import { ApifyClient as _ApifyClient } from 'apify-client';
 import type { AxiosRequestConfig } from 'axios';
 
 import { USER_AGENT_ORIGIN } from './const.js';
+import type { AuthToken } from './types.js';
 
 /**
  * Adds a User-Agent header to the request config.
@@ -22,23 +23,59 @@ export function getApifyAPIBaseUrl(): string {
     return process.env.APIFY_API_BASE_URL || 'https://api.apify.com';
 }
 
+/**
+ * Adds Skyfire header to the request config if needed.
+ * @param config
+ * @param authToken
+ * @private
+ */
+function addSkyfireHeader(config: AxiosRequestConfig, authToken?: AuthToken): AxiosRequestConfig {
+    if (authToken?.type === 'skyfire') {
+        const updatedConfig = { ...config };
+        updatedConfig.headers = updatedConfig.headers ?? {};
+        updatedConfig.headers['skyfire-pay-id'] = authToken.value;
+        return updatedConfig;
+    }
+    return config;
+}
+
 export class ApifyClient extends _ApifyClient {
-    constructor(options: ApifyClientOptions) {
+    constructor(options: ApifyClientOptions & { authToken?: AuthToken }) {
+        // Destructure to separate authToken from other options
+        const { authToken, ...clientOptions } = options;
+
         /**
          * In order to publish to DockerHub, we need to run their build task to validate our MCP server.
          * This was failing since we were sending this dummy token to Apify in order to build the Actor tools.
          * So if we encounter this dummy value, we remove it to use Apify client as unauthenticated, which is sufficient
          * for server start and listing of tools.
          */
-        if (options.token?.toLowerCase() === 'your-apify-token') {
-            // eslint-disable-next-line no-param-reassign
-            delete options.token;
+        if (clientOptions.token?.toLowerCase() === 'your-apify-token') {
+            delete clientOptions.token;
+        }
+
+        // Handle authToken if provided
+        if (authToken) {
+            if (authToken.type === 'skyfire') {
+                // For Skyfire tokens: DO NOT set as bearer token
+                // Only add the skyfire-pay-id header via request interceptor
+                // Remove any existing token to ensure no bearer auth
+                delete clientOptions.token;
+            } else {
+                // For Apify tokens: Use as regular bearer token (existing behavior)
+                clientOptions.token = authToken.value;
+            }
+        }
+
+        const requestInterceptors = [addUserAgent];
+        if (authToken?.type === 'skyfire') {
+            requestInterceptors.push((config) => addSkyfireHeader(config, authToken));
         }
 
         super({
-            ...options,
+            ...clientOptions, // Now safe to spread without authToken
             baseUrl: getApifyAPIBaseUrl(),
-            requestInterceptors: [addUserAgent],
+            requestInterceptors,
         });
     }
 }

src/main.ts

@@ -11,7 +11,7 @@ import log from '@apify/log';
 import { createExpressApp } from './actor/server.js';
 import { processInput } from './input.js';
 import { callActorGetDataset } from './tools/index.js';
-import type { Input } from './types.js';
+import type { AuthToken, Input } from './types.js';
 
 const STANDBY_MODE = Actor.getEnv().metaOrigin === 'STANDBY';
 
@@ -25,6 +25,11 @@ if (!process.env.APIFY_TOKEN) {
     process.exit(1);
 }
 
+const authToken: AuthToken = {
+    value: process.env.APIFY_TOKEN,
+    type: 'apify',
+};
+
 const input = processInput((await Actor.getInput<Partial<Input>>()) ?? ({} as Input));
 log.info('Loaded input', { input: JSON.stringify(input) });
 
@@ -44,7 +49,7 @@ if (STANDBY_MODE) {
         await Actor.fail('If you need to debug a specific Actor, please provide the debugActor and debugActorInput fields in the input');
     }
     const options = { memory: input.maxActorMemoryBytes } as ActorCallOptions;
-    const { items } = await callActorGetDataset(input.debugActor!, input.debugActorInput!, process.env.APIFY_TOKEN, options);
+    const { items } = await callActorGetDataset(input.debugActor!, input.debugActorInput!, authToken, options);
 
     await Actor.pushData(items);
     log.info('Pushed items to dataset', { itemCount: items.count });

src/mcp/actors.ts

@@ -1,6 +1,5 @@
 import type { ActorDefinition } from 'apify-client';
 
-import { ApifyClient } from '../apify-client.js';
 import { MCP_STREAMABLE_ENDPOINT } from '../const.js';
 import type { ActorDefinitionPruned } from '../types.js';
 
@@ -44,38 +43,8 @@ export async function getActorMCPServerURL(realActorId: string, mcpServerPath: s
 }
 
 /**
-* Gets Actor ID from the Actor object.
-*/
-export async function getRealActorID(actorIdOrName: string, apifyToken: string): Promise<string> {
-    const apifyClient = new ApifyClient({ token: apifyToken });
-
-    const actor = apifyClient.actor(actorIdOrName);
-    const info = await actor.get();
-    if (!info) {
-        throw new Error(`Actor ${actorIdOrName} not found`);
-    }
-    return info.id;
-}
-
-/**
-* Returns standby URL for given Actor ID.
-*/
+ * Returns standby URL for given Actor ID.
+ */
 export async function getActorStandbyURL(realActorId: string, standbyBaseUrl = 'apify.actor'): Promise<string> {
     return `https://${realActorId}.${standbyBaseUrl}`;
 }
-
-export async function getActorDefinition(actorID: string, apifyToken: string): Promise<ActorDefinition> {
-    const apifyClient = new ApifyClient({ token: apifyToken });
-    const actor = apifyClient.actor(actorID);
-    const defaultBuildClient = await actor.defaultBuild();
-    const buildInfo = await defaultBuildClient.get();
-    if (!buildInfo) {
-        throw new Error(`Default build for Actor ${actorID} not found`);
-    }
-    const { actorDefinition } = buildInfo;
-    if (!actorDefinition) {
-        throw new Error(`Actor default build ${actorID} does not have Actor definition`);
-    }
-
-    return actorDefinition;
-}

src/mcp/server.ts

@@ -30,7 +30,7 @@ import {
 import { prompts } from '../prompts/index.js';
 import { callActorGetDataset, defaultTools, getActorsAsTools, toolCategories } from '../tools/index.js';
 import { decodeDotPropertyNames } from '../tools/utils.js';
-import type { ActorMcpTool, ActorTool, HelperTool, ToolEntry } from '../types.js';
+import type { ActorMcpTool, ActorTool, AuthToken, HelperTool, ToolEntry } from '../types.js';
 import { createProgressTracker } from '../utils/progress.js';
 import { getToolPublicFieldOnly } from '../utils/tools.js';
 import { connectMCPClient } from './client.js';
@@ -160,7 +160,7 @@ export class ActorsMcpServer {
     * @param toolNames - Array of tool names to ensure are loaded
     * @param apifyToken - Apify API token for authentication
     */
-    public async loadToolsByName(toolNames: string[], apifyToken: string) {
+    public async loadToolsByName(toolNames: string[], authToken: AuthToken) {
         const loadedTools = this.listAllToolNames();
         const actorsToLoad: string[] = [];
         const toolsToLoad: ToolEntry[] = [];
@@ -185,7 +185,7 @@ export class ActorsMcpServer {
         }
 
         if (actorsToLoad.length > 0) {
-            await this.loadActorsAsTools(actorsToLoad, apifyToken);
+            await this.loadActorsAsTools(actorsToLoad, authToken);
         }
     }
 
@@ -196,8 +196,8 @@ export class ActorsMcpServer {
      * @param apifyToken - Apify API token for authentication
      * @returns Promise<ToolEntry[]> - Array of loaded tool entries
      */
-    public async loadActorsAsTools(actorIdsOrNames: string[], apifyToken: string): Promise<ToolEntry[]> {
-        const actorTools = await getActorsAsTools(actorIdsOrNames, apifyToken);
+    public async loadActorsAsTools(actorIdsOrNames: string[], authToken: AuthToken): Promise<ToolEntry[]> {
+        const actorTools = await getActorsAsTools(actorIdsOrNames, authToken);
         if (actorTools.length > 0) {
             this.upsertTools(actorTools, true);
         }
@@ -211,8 +211,8 @@ export class ActorsMcpServer {
      *
      * Used primarily for SSE.
      */
-    public async loadToolsFromUrl(url: string, apifyToken: string) {
-        const tools = await processParamsGetTools(url, apifyToken);
+    public async loadToolsFromUrl(url: string, authToken: AuthToken) {
+        const tools = await processParamsGetTools(url, authToken);
         if (tools.length > 0) {
             log.debug('Loading tools from query parameters');
             this.upsertTools(tools, false);
@@ -381,17 +381,28 @@ export class ActorsMcpServer {
             // eslint-disable-next-line prefer-const
             let { name, arguments: args, _meta: meta } = request.params;
             const { progressToken } = meta || {};
-            const apifyToken = (request.params.apifyToken || process.env.APIFY_TOKEN) as string;
+            // Extract auth token with fallback to APIFY_TOKEN environment variable
+            let authToken: AuthToken | undefined = request.params.authToken as AuthToken;
+
+            // Fallback to APIFY_TOKEN environment variable for local development
+            if (!authToken && process.env.APIFY_TOKEN) {
+                authToken = {
+                    value: process.env.APIFY_TOKEN,
+                    type: 'apify', // Environment variable is always an Apify token
+                };
+            }
             const userRentedActorIds = request.params.userRentedActorIds as string[] | undefined;
 
-            // Remove apifyToken from request.params just in case
-            delete request.params.apifyToken;
+            // Remove authToken from request.params only if it was provided in params
+            if (request.params.authToken) {
+                delete request.params.authToken;
+            }
             // Remove other custom params passed from apify-mcp-server
             delete request.params.userRentedActorIds;
 
-            // Validate token
-            if (!apifyToken) {
-                const msg = 'APIFY_TOKEN is required. It must be set in the environment variables or passed as a parameter in the body.';
+            // Validate auth token
+            if (!authToken || !authToken.value) {
+                const msg = 'Valid authentication token required. It must be provided either in the authToken parameter or APIFY_TOKEN environment variable.';
                 log.error(msg);
                 await this.server.sendLoggingMessage({ level: 'error', data: msg });
                 throw new McpError(
@@ -462,7 +473,7 @@ export class ActorsMcpServer {
                         extra,
                         apifyMcpServer: this,
                         mcpServer: this.server,
-                        apifyToken,
+                        authToken,
                         userRentedActorIds,
                         progressTracker,
                     }) as object;
@@ -478,7 +489,7 @@ export class ActorsMcpServer {
                     const serverTool = tool.tool as ActorMcpTool;
                     let client: Client | undefined;
                     try {
-                        client = await connectMCPClient(serverTool.serverUrl, apifyToken);
+                        client = await connectMCPClient(serverTool.serverUrl, authToken.value);
 
                         // Only set up notification handlers if progressToken is provided by the client
                         if (progressToken) {
@@ -527,7 +538,7 @@ export class ActorsMcpServer {
                         const { runId, datasetId, items } = await callActorGetDataset(
                             actorTool.actorFullName,
                             args,
-                            apifyToken as string,
+                            authToken,
                             callOptions,
                             progressTracker,
                         );

src/mcp/utils.ts

@@ -2,7 +2,7 @@ import { createHash } from 'node:crypto';
 import { parse } from 'node:querystring';
 
 import { processInput } from '../input.js';
-import type { Input } from '../types.js';
+import type { AuthToken, Input } from '../types.js';
 import { loadToolsFromInput } from '../utils/tools-loader.js';
 import { MAX_TOOL_NAME_LENGTH, SERVER_ID_LENGTH } from './const.js';
 
@@ -37,11 +37,11 @@ export function getProxyMCPServerToolName(url: string, toolName: string): string
  * Process input parameters from URL and get tools
  * If URL contains query parameter `actors`, return tools from Actors otherwise return null.
  * @param url
- * @param apifyToken
+ * @param authToken
  */
-export async function processParamsGetTools(url: string, apifyToken: string) {
+export async function processParamsGetTools(url: string, authToken: AuthToken) {
     const input = parseInputParamsFromUrl(url);
-    return await loadToolsFromInput(input, apifyToken);
+    return await loadToolsFromInput(input, authToken);
 }
 
 export function parseInputParamsFromUrl(url: string): Input {

src/stdio.ts

@@ -24,7 +24,7 @@ import log from '@apify/log';
 
 import { processInput } from './input.js';
 import { ActorsMcpServer } from './mcp/server.js';
-import type { Input, ToolSelector } from './types.js';
+import type { AuthToken, Input, ToolSelector } from './types.js';
 import { loadToolsFromInput } from './utils/tools-loader.js';
 
 // Keeping this interface here and not types.ts since
@@ -122,7 +122,11 @@ async function main() {
     const normalized = processInput(input);
 
     // Use the shared tools loading logic
-    const tools = await loadToolsFromInput(normalized, process.env.APIFY_TOKEN as string);
+    const authToken: AuthToken = {
+        value: process.env.APIFY_TOKEN as string,
+        type: 'apify',
+    };
+    const tools = await loadToolsFromInput(normalized, authToken);
 
     mcpServer.upsertTools(tools);