How to chain multiple MCP tools in a workflow

1// Pattern 1: AI orchestration (natural)
2// You just ask: "Query the orders table for this month, export as CSV,
3// and notify the team on Slack"
4// The AI calls: query_db → write_file → send_slack_message
5
6// Pattern 2: Scripted client orchestration
7const dbResult = await client.callTool({ name: "query_db", arguments: { sql: "SELECT ..." } });
8const csvPath = await client.callTool({ name: "write_csv", arguments: { data: dbResult, path: "report.csv" } });
9await client.callTool({ name: "send_notification", arguments: { message: `Report ready: ${csvPath}` } });
10
11// Pattern 3: Composite server tool
12server.tool("generate_and_notify_report", "...", schema, async (params) => {
13  const data = await queryDatabase(params.sql);
14  const path = await writeCsv(data, params.filename);
15  await sendNotification(`Report ready: ${path}`);
16  return { content: [{ type: "text", text: `Report generated and team notified` }] };
17});

1// src/workflow.ts
2import { Client } from "@modelcontextprotocol/sdk/client/index.js";
3import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";
4
5async function connectServer(command: string, args: string[]): Promise<Client> {
6  const transport = new StdioClientTransport({ command, args });
7  const client = new Client({ name: "workflow", version: "1.0.0" }, { capabilities: {} });
8  await client.connect(transport);
9  return client;
10}
11
12function extractText(result: any): string {
13  return (result.content as any[])
14    .filter(c => c.type === "text")
15    .map(c => c.text)
16    .join("\n");
17}
18
19async function runPipeline() {
20  // Connect to servers
21  const dbClient = await connectServer("node", ["dist/db-server.js"]);
22  const fileClient = await connectServer("node", ["dist/file-server.js"]);
23
24  try {
25    // Step 1: Query database
26    console.error("[pipeline] Step 1: Querying database...");
27    const queryResult = await dbClient.callTool({
28      name: "query",
29      arguments: { sql: "SELECT * FROM orders WHERE created_at >= NOW() - INTERVAL '7 days'" },
30    });
31    if ((queryResult as any).isError) throw new Error(`Query failed: ${extractText(queryResult)}`);
32    const data = JSON.parse(extractText(queryResult));
33    console.error(`[pipeline] Got ${data.rows} rows`);
34
35    // Step 2: Write CSV report
36    console.error("[pipeline] Step 2: Writing CSV...");
37    const csvResult = await fileClient.callTool({
38      name: "write_file",
39      arguments: {
40        filePath: "reports/weekly-orders.csv",
41        content: formatAsCsv(data),
42      },
43    });
44    if ((csvResult as any).isError) throw new Error(`CSV write failed: ${extractText(csvResult)}`);
45
46    // Step 3: Write summary
47    console.error("[pipeline] Step 3: Writing summary...");
48    const summary = `Weekly Orders Report\nGenerated: ${new Date().toISOString()}\nTotal rows: ${data.rows}\nTotal revenue: $${data.data.reduce((s: number, r: any) => s + (r.total || 0), 0).toFixed(2)}`;
49    await fileClient.callTool({
50      name: "write_file",
51      arguments: { filePath: "reports/weekly-summary.txt", content: summary },
52    });
53
54    console.error("[pipeline] Complete!");
55  } finally {
56    await dbClient.close();
57    await fileClient.close();
58  }
59}
60
61function formatAsCsv(data: { columns: string[]; data: any[] }): string {
62  const header = data.columns.join(",");
63  const rows = data.data.map((row: any) =>
64    data.columns.map(col => JSON.stringify(row[col] ?? "")).join(",")
65  );
66  return [header, ...rows].join("\n");
67}
68
69runPipeline().catch(e => { console.error("Pipeline failed:", e); process.exit(1); });

1// src/workflow-server.ts
2import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
3import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
4import { z } from "zod";
5import pg from "pg";
6import fs from "fs/promises";
7
8const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL });
9const server = new McpServer({ name: "workflow-server", version: "1.0.0" });
10
11server.tool(
12  "generate_report",
13  "Query data, generate CSV, and create summary report in one step",
14  {
15    query: z.string().describe("SQL SELECT query for the report data"),
16    reportName: z.string().describe("Name for the report files (no extension)"),
17    outputDir: z.string().default("/tmp/reports"),
18  },
19  async ({ query, reportName, outputDir }) => {
20    const steps: string[] = [];
21
22    try {
23      // Step 1: Execute query
24      if (/\b(DELETE|DROP|UPDATE|INSERT)\b/i.test(query)) {
25        return { content: [{ type: "text", text: "Error: Only SELECT queries allowed" }], isError: true };
26      }
27      const result = await pool.query({ text: query, timeout: 30000 });
28      steps.push(`Queried ${result.rowCount} rows`);
29
30      // Step 2: Generate CSV
31      await fs.mkdir(outputDir, { recursive: true });
32      const csvPath = `${outputDir}/${reportName}.csv`;
33      const header = result.fields.map(f => f.name).join(",");
34      const rows = result.rows.map(r =>
35        result.fields.map(f => JSON.stringify(r[f.name] ?? "")).join(",")
36      );
37      await fs.writeFile(csvPath, [header, ...rows].join("\n"));
38      steps.push(`CSV written to ${csvPath}`);
39
40      // Step 3: Generate summary
41      const summaryPath = `${outputDir}/${reportName}-summary.json`;
42      const summary = {
43        reportName,
44        generatedAt: new Date().toISOString(),
45        rowCount: result.rowCount,
46        columns: result.fields.map(f => f.name),
47        csvPath,
48      };
49      await fs.writeFile(summaryPath, JSON.stringify(summary, null, 2));
50      steps.push(`Summary written to ${summaryPath}`);
51
52      return {
53        content: [{ type: "text", text: `Report pipeline complete:\n${steps.map((s, i) => `  ${i + 1}. ${s}`).join("\n")}` }],
54      };
55    } catch (error) {
56      return {
57        content: [{
58          type: "text",
59          text: `Pipeline failed at step ${steps.length + 1}:\n` +
60            `Completed: ${steps.join(", ") || "none"}\n` +
61            `Error: ${error instanceof Error ? error.message : String(error)}`,
62        }],
63        isError: true,
64      };
65    }
66  }
67);

1// src/safe-pipeline.ts
2interface PipelineStep<T> {
3  name: string;
4  execute: (context: T) => Promise<T>;
5  rollback?: (context: T) => Promise<void>;
6}
7
8async function runPipeline<T>(steps: PipelineStep<T>[], initialContext: T): Promise<T> {
9  const completed: PipelineStep<T>[] = [];
10  let context = initialContext;
11
12  for (const step of steps) {
13    try {
14      console.error(`[pipeline] Running: ${step.name}`);
15      context = await step.execute(context);
16      completed.push(step);
17      console.error(`[pipeline] Completed: ${step.name}`);
18    } catch (error) {
19      console.error(`[pipeline] Failed: ${step.name} - ${error}`);
20
21      // Rollback completed steps in reverse order
22      for (const completedStep of completed.reverse()) {
23        if (completedStep.rollback) {
24          try {
25            console.error(`[pipeline] Rolling back: ${completedStep.name}`);
26            await completedStep.rollback(context);
27          } catch (rollbackError) {
28            console.error(`[pipeline] Rollback failed: ${completedStep.name} - ${rollbackError}`);
29          }
30        }
31      }
32
33      throw new Error(
34        `Pipeline failed at "${step.name}": ${error instanceof Error ? error.message : String(error)}\n` +
35        `Completed steps: ${completed.map(s => s.name).join(", ") || "none"}\n` +
36        `Rolled back: ${completed.filter(s => s.rollback).map(s => s.name).join(", ") || "none"}`
37      );
38    }
39  }
40
41  return context;
42}
43
44// Usage:
45// const result = await runPipeline([
46//   { name: "query", execute: queryStep },
47//   { name: "write-csv", execute: writeCsvStep, rollback: deleteFileStep },
48//   { name: "notify", execute: notifyStep },
49// ], { sql: "SELECT ...", outputDir: "/tmp" });

1// Example: Incident investigation pipeline
2// Chains: Sentry → Database → Filesystem → Linear
3
4async function investigateIncident(project: string) {
5  const sentry = await connectServer("node", ["dist/sentry-server.js"]);
6  const db = await connectServer("node", ["dist/db-server.js"]);
7  const files = await connectServer("npx", ["-y", "@modelcontextprotocol/server-filesystem", "."]);
8  const linear = await connectServer("node", ["dist/linear-server.js"]);
9
10  try {
11    // Step 1: Get recent errors from Sentry
12    const errors = await sentry.callTool({
13      name: "recent_errors", arguments: { project, hours: 1 },
14    });
15    const topError = JSON.parse(extractText(errors))[0];
16
17    // Step 2: Get error details and stack trace
18    const details = await sentry.callTool({
19      name: "error_details", arguments: { issueId: topError.id },
20    });
21    const errorInfo = JSON.parse(extractText(details));
22
23    // Step 3: Query related database records
24    const dbResult = await db.callTool({
25      name: "query", arguments: {
26        sql: `SELECT * FROM error_logs WHERE message LIKE '%${topError.title.slice(0, 50)}%' LIMIT 5`,
27      },
28    });
29
30    // Step 4: Read the failing source file
31    const frame = errorInfo.frames?.[0];
32    const sourceCode = frame ? await files.callTool({
33      name: "read_file", arguments: { path: frame.file },
34    }) : null;
35
36    // Step 5: Create a Linear ticket with all context
37    await linear.callTool({
38      name: "create_issue",
39      arguments: {
40        title: `[Auto] ${topError.title}`,
41        description: `Error count: ${topError.count}\nStack: ${frame?.file}:${frame?.line}\nLast seen: ${topError.lastSeen}`,
42        teamKey: "ENG",
43        priority: topError.count > 100 ? 1 : 2,
44      },
45    });
46
47    console.error("Incident investigated and ticket created");
48  } finally {
49    await Promise.all([sentry.close(), db.close(), files.close(), linear.close()]);
50  }
51}