MCP Tools

An MCP tool is a class decorated with @McpTool that implements McpToolHandler. The decorator describes what the client sees; the class implements a single handle method.

import { z } from 'zod';
import {
  McpFlow,
  McpTool,
  McpToolHandler,
  McpToolResult,
  mcpResult,
} from '@picsart/pa-pluggable-workers-core';

@McpTool({
  name: 'remove_background',
  description: 'Removes the background from an image and returns a CDN URL.',
  inputSchema: {
    imageUrl: z.string().describe('Publicly reachable source image URL'),
  },
})
export class RemoveBackgroundMcpTool implements McpToolHandler {

  async handle(args: { imageUrl: string }, flow: McpFlow): Promise<McpToolResult> {
    const res = await flow.executeTask('REMOVE_BG', { imageUrl: args.imageUrl });
    return mcpResult.json(res.result);
  }
}

Register the class as a provider in your CoreModule, exactly like an executor:

@Module({
  providers: [RemoveBackgroundMcpTool],
})
export class CoreModule {}

Definition

Field	Required	Description
name	✅	Tool id shown to clients (e.g. remove_background). Unique per worker.
description	✅	Human / model facing description shown in the tool list.
inputSchema		Argument name → Zod type. Use .describe() to document each argument for the model. Defaults to no arguments.
annotations		Optional MCP annotations (title, readOnlyHint, …).

Returning a result

Build the result with the mcpResult helpers rather than assembling it by hand:

Helper	Returns
mcpResult.json(data)	data as JSON (also echoed as structured content).
mcpResult.text(str)	A plain text result.
mcpResult.error(msg)	An error result.

A thrown error is turned into an error result automatically — use mcpResult.error only when you want to return a controlled failure message without throwing.

The flow

The handle method receives an McpFlow for orchestrating your existing tasks:

• flow.executeTask(command, params, options?) — run a task you've already built and use its result. options accepts { timeout, config }.
• flow.streamTask(command, params, options?) — run a streaming task and iterate its events as they arrive. options accepts { config }. The final task.completed event carries the result; a failed task throws instead of yielding task.failed.
• flow.emitProgress({ progress, total?, message? }) — stream a progress update to the client while you work. This is best-effort: it is shown only if the client supports progress, so it never breaks clients that don't.
• flow.meta() — read request metadata (see Flow Metadata).

await flow.emitProgress({ progress: 0, total: 1, message: 'Starting' });
const res = await flow.executeTask('REMOVE_BG', { imageUrl: args.imageUrl });
await flow.emitProgress({ progress: 1, total: 1, message: 'Done' });

Relaying a streaming task's progress to the client:

let result;
for await (const event of flow.streamTask('GENERATE_VIDEO', { prompt: args.prompt }) as AsyncGenerator<any>) {
    if (event.type === EventTypes.METRICS) {
        await flow.emitProgress({ progress: event.percent, total: 100 });
    }
    if (event.type === EventTypes.COMPLETED) {
        result = event.result;
    }
}

MCP call options don't accept configId — pass the config inline with config.

Sub-tasks are charged automatically

Tasks a tool runs via flow.executeTask or flow.streamTask are always charged for credits — this cannot be disabled for MCP tools. The tool itself is not billed; only its sub-tasks are. Every task you compose must support credit management (see Monetization → Sub-Task Charging), otherwise the call fails.