feat: call-actor can be cancelled

Author: MichalKalita

src/main.ts

@@ -44,10 +44,12 @@ 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 result = await callActorGetDataset(input.debugActor!, input.debugActorInput!, process.env.APIFY_TOKEN, options);
 
-    await Actor.pushData(items);
-    log.info('Pushed items to dataset', { itemCount: items.count });
+    if (result && result.items) {
+        await Actor.pushData(result.items);
+        log.info('Pushed items to dataset', { itemCount: result.items.count });
+    }
     await Actor.exit();
 }
 

src/mcp/server.ts

@@ -534,7 +534,8 @@ export class ActorsMcpServer {
                         );
 
                         if (!result) {
-                            // If the actor was aborted by the client, we don't want to return anything
+                            // Receivers of cancellation notifications SHOULD NOT send a response for the cancelled request
+                            // https://modelcontextprotocol.io/specification/2025-06-18/basic/utilities/cancellation#behavior-requirements
                             return { };
                         }
 
@@ -549,12 +550,6 @@ export class ActorsMcpServer {
                         });
                         content.push(...itemContents);
                         return { content };
-                    } catch (error) {
-                        if (error instanceof Error && error.message === 'Operation cancelled') {
-                            // Receivers of cancellation notifications SHOULD NOT send a response for the cancelled request
-                            // https://modelcontextprotocol.io/specification/2025-06-18/basic/utilities/cancellation#behavior-requirements
-                            return { };
-                        }
                     } finally {
                         if (progressTracker) {
                             progressTracker.stop();

src/tools/actor.ts

@@ -49,21 +49,6 @@ export type CallActorGetDatasetResult = {
  * @returns {Promise<CallActorGetDatasetResult | null>} - A promise that resolves to an object containing the actor run and dataset items.
  * @throws {Error} - Throws an error if the `APIFY_TOKEN` is not set
  */
-export async function callActorGetDataset(
-    actorName: string,
-    input: unknown,
-    apifyToken: string,
-    callOptions: ActorCallOptions | undefined,
-    progressTracker?: ProgressTracker | null,
-): Promise<CallActorGetDatasetResult>; // Without abort signal Result or Error is returned
-export async function callActorGetDataset(
-    actorName: string,
-    input: unknown,
-    apifyToken: string,
-    callOptions: ActorCallOptions | undefined,
-    progressTracker: ProgressTracker | null,
-    abortSignal: AbortSignal,
-): Promise<CallActorGetDatasetResult | null>; // With abort signal, null is returned if the actor was aborted by the client
 export async function callActorGetDataset(
     actorName: string,
     input: unknown,
@@ -90,7 +75,7 @@ export async function callActorGetDataset(
             abortSignal?.addEventListener('abort', async () => {
                 // Abort the actor run via API
                 try {
-                    await client.run(actorRun.id).abort({ gracefully: true });
+                    await client.run(actorRun.id).abort({ gracefully: false });
                 } catch (e) {
                     log.error('Error aborting Actor run', { error: e, runId: actorRun.id });
                 }
@@ -335,7 +320,7 @@ The step parameter enforces this workflow - you cannot call an Actor without fir
         inputSchema: zodToJsonSchema(callActorArgs),
         ajvValidate: ajv.compile(zodToJsonSchema(callActorArgs)),
         call: async (toolArgs) => {
-            const { args, apifyToken, progressTracker } = toolArgs;
+            const { args, apifyToken, progressTracker, extra } = toolArgs;
             const { actor: actorName, step, input, callOptions } = callActorArgs.parse(args);
 
             try {
@@ -384,14 +369,23 @@ The step parameter enforces this workflow - you cannot call an Actor without fir
                     }
                 }
 
-                const { runId, datasetId, items } = await callActorGetDataset(
+                const result = await callActorGetDataset(
                     actorName,
                     input,
                     apifyToken,
                     callOptions,
                     progressTracker,
+                    extra.signal,
                 );
 
+                if (!result) {
+                    // Receivers of cancellation notifications SHOULD NOT send a response for the cancelled request
+                    // https://modelcontextprotocol.io/specification/2025-06-18/basic/utilities/cancellation#behavior-requirements
+                    return { };
+                }
+
+                const { runId, datasetId, items } = result;
+
                 const content = [
                     { type: 'text', text: `Actor finished with runId: ${runId}, datasetId ${datasetId}` },
                 ];

tests/integration/suite.ts

@@ -631,8 +631,9 @@ export function createIntegrationTestsSuite(
         });
 
         // Cancellation test: start a long-running actor and cancel immediately, then verify it was aborted
-        it('should abort actor run when request is cancelled', async () => {
-            const ACTOR_NAME = 'michal.kalita/test-timeout';
+        // Is not possible to run this test in parallel
+        it.runIf(options.transport === 'streamable-http')('should abort actor run on notifications/cancelled', async () => {
+            const ACTOR_NAME = 'apify/rag-web-browser';
             const selectedToolName = actorNameToToolName(ACTOR_NAME);
             const client = await createClientFn({ enableAddingActors: true });
 
@@ -646,7 +647,7 @@ export function createIntegrationTestsSuite(
                 method: 'tools/call' as const,
                 params: {
                     name: selectedToolName,
-                    arguments: { timeout: 30 },
+                    arguments: { query: 'restaurants in San Francisco', maxResults: 10 },
                 },
             }, CallToolResultSchema, { signal: controller.signal })
             // Ignores error "AbortError: This operation was aborted"
@@ -672,7 +673,52 @@ export function createIntegrationTestsSuite(
                     return run.status === 'ABORTED' || run.status === 'ABORTING';
                 }
                 return false;
-            }, { timeout: 30000, interval: 1000 });
+            }, { timeout: 3000, interval: 500 });
+        });
+
+        // Cancellation test using call-actor tool: start a long-running actor via call-actor and cancel immediately, then verify it was aborted
+        it.runIf(options.transport === 'streamable-http')('should abort call-actor tool on notifications/cancelled', async () => {
+            const ACTOR_NAME = 'apify/rag-web-browser';
+            const client = await createClientFn({ tools: ['actors'] });
+
+            // Build request and cancel immediately via AbortController
+            const controller = new AbortController();
+
+            const requestPromise = client.request({
+                method: 'tools/call' as const,
+                params: {
+                    name: HelperTools.ACTOR_CALL,
+                    arguments: {
+                        actor: ACTOR_NAME,
+                        step: 'call',
+                        input: { query: 'restaurants in San Francisco', maxResults: 10 },
+                    },
+                },
+            }, CallToolResultSchema, { signal: controller.signal })
+            // Ignores error "AbortError: This operation was aborted"
+                .catch(() => undefined);
+
+            // Abort right away
+            setTimeout(() => controller.abort(), 1000);
+
+            // Ensure the request completes/cancels before proceeding
+            await requestPromise;
+
+            // Verify via Apify API that a recent run for this actor was aborted
+            const api = new ApifyClient({ token: process.env.APIFY_TOKEN as string });
+            const actor = await api.actor(ACTOR_NAME).get();
+            expect(actor).toBeDefined();
+            const actId = actor!.id as string;
+
+            // Poll up to 30s for the latest run for this actor to reach ABORTED/ABORTING
+            await vi.waitUntil(async () => {
+                const runsList = await api.runs().list({ limit: 5, desc: true });
+                const run = runsList.items.find((r) => r.actId === actId);
+                if (run) {
+                    return run.status === 'ABORTED' || run.status === 'ABORTING';
+                }
+                return false;
+            }, { timeout: 3000, interval: 500 });
         });
 
         // Environment variable tests - only applicable to stdio transport