Add PipelineError for plugin pipeline failures (#22)

Adds support for detecting and handling mcpd plugin pipeline failures:

- Add PipelineError class with pipelineFlow, serverName, and operation properties
- Detect Mcpd-Error-Type header on 500 responses (request-pipeline-failure, response-pipeline-failure)
- Enrich PipelineError with server/tool context when called through performCall
- Export PipelineFlow type and PIPELINE_FLOW_REQUEST/PIPELINE_FLOW_RESPONSE constants
- Add comprehensive tests for pipeline error handling
- Update README with PipelineError documentation

Author: peteski22

README.md

@@ -609,12 +609,20 @@ import {
   ToolExecutionError, // Tool execution failed
   ValidationError, // Input validation failed
   TimeoutError, // Operation timed out
+  PipelineError, // Pipeline processing failed
 } from "@mozilla-ai/mcpd";
 
 try {
   const result = await client.servers.unknown.tools.tool();
 } catch (error) {
-  if (error instanceof ToolNotFoundError) {
+  if (error instanceof PipelineError) {
+    // Pipeline failure - a required plugin failed during processing.
+    console.error(`Pipeline ${error.pipelineFlow} failure:`, error.message);
+    if (error.pipelineFlow === "response") {
+      // Tool was called but results cannot be delivered.
+      console.error("A required plugin failed during response processing");
+    }
+  } else if (error instanceof ToolNotFoundError) {
     console.error(
       `Tool not found: ${error.toolName} on server ${error.serverName}`,
     );
@@ -626,6 +634,20 @@ try {
 }
 ```
 
+### PipelineError
+
+The `PipelineError` is thrown when mcpd's plugin pipeline fails. This indicates a problem with a plugin or an external system that a plugin depends on (e.g., audit service, authentication provider), not a problem with your request or the tool itself.
+
+- **Response Pipeline Failure** (`pipelineFlow === 'response'`): The upstream request was processed (the tool was called), but results cannot be returned because a required plugin (like audit logging) failed during response processing. This does not indicate whether the tool itself succeeded or failed.
+- **Request Pipeline Failure** (`pipelineFlow === 'request'`): The request was rejected before reaching the upstream server because a required plugin (like authentication) failed during request processing.
+
+Properties:
+
+- `message: string` - Descriptive error message
+- `serverName?: string` - The server name (when called through tool execution)
+- `operation?: string` - The operation (e.g., "time.get_current_time")
+- `pipelineFlow?: 'request' | 'response'` - Which pipeline flow failed
+
 ## Development
 
 ### Setup

src/client.ts

@@ -19,6 +19,10 @@ import {
   ServerUnhealthyError,
   ToolExecutionError,
   TimeoutError,
+  PipelineError,
+  PIPELINE_FLOW_REQUEST,
+  PIPELINE_FLOW_RESPONSE,
+  type PipelineFlow,
 } from "./errors";
 import {
   HealthStatusHelpers,
@@ -74,6 +78,19 @@ const SERVER_HEALTH_CACHE_MAXSIZE = 100;
  */
 const TOOL_SEPARATOR = "__";
 
+/**
+ * Header name for mcpd pipeline error type.
+ */
+const MCPD_ERROR_TYPE_HEADER = "Mcpd-Error-Type";
+
+/**
+ * Maps mcpd error type header values to pipeline flows.
+ */
+const PIPELINE_ERROR_FLOWS: Record<string, PipelineFlow> = {
+  "request-pipeline-failure": PIPELINE_FLOW_REQUEST,
+  "response-pipeline-failure": PIPELINE_FLOW_RESPONSE,
+};
+
 /**
  * Type alias for agent functions in array format.
  * @internal
@@ -236,6 +253,26 @@ export class McpdClient {
           errorModel = null;
         }
 
+        // Check for pipeline failure (500 with Mcpd-Error-Type header).
+        if (response.status === 500) {
+          const errorType = response.headers
+            .get(MCPD_ERROR_TYPE_HEADER)
+            ?.toLowerCase();
+
+          const flow = errorType ? PIPELINE_ERROR_FLOWS[errorType] : undefined;
+
+          if (flow) {
+            const message = errorModel?.detail || body || "Pipeline failure";
+
+            throw new PipelineError(
+              message,
+              undefined, // serverName - enriched by caller if available.
+              undefined, // operation - enriched by caller if available.
+              flow,
+            );
+          }
+        }
+
         if (errorModel && errorModel.detail) {
           const errorDetails = errorModel.errors
             ?.map((e) => `${e.location}: ${e.message}`)
@@ -803,6 +840,17 @@ export class McpdClient {
       // Return the response as-is (already parsed or not a string)
       return response;
     } catch (error) {
+      // Enrich PipelineError with server/tool context.
+      if (error instanceof PipelineError) {
+        throw new PipelineError(
+          error.message,
+          serverName,
+          `${serverName}.${toolName}`,
+          error.pipelineFlow,
+          error.cause as Error | undefined,
+        );
+      }
+
       if (error instanceof McpdError) {
         throw error;
       }

src/errors.ts

@@ -7,6 +7,23 @@
 
 import type { ErrorModel } from "./types";
 
+/**
+ * Pipeline flow constant for request processing failures.
+ */
+export const PIPELINE_FLOW_REQUEST = "request" as const;
+
+/**
+ * Pipeline flow constant for response processing failures.
+ */
+export const PIPELINE_FLOW_RESPONSE = "response" as const;
+
+/**
+ * Pipeline flow indicating where in the pipeline the failure occurred.
+ */
+export type PipelineFlow =
+  | typeof PIPELINE_FLOW_REQUEST
+  | typeof PIPELINE_FLOW_RESPONSE;
+
 /**
  * Base exception for all mcpd SDK errors.
  *
@@ -201,3 +218,74 @@ export class TimeoutError extends McpdError {
     Error.captureStackTrace(this, this.constructor);
   }
 }
+
+/**
+ * Raised when required pipeline processing fails.
+ *
+ * This indicates that required processing failed in the mcpd pipeline.
+ * The error occurs when a required plugin (such as authentication, validation,
+ * audit logging, monitoring, or response transformation) fails during request
+ * or response processing.
+ *
+ * Pipeline Flow Distinction:
+ * - **response-pipeline-failure**: The upstream request was processed (the tool
+ *   was called), but results cannot be returned due to a required response
+ *   processing step failure. Note: This does not indicate whether the tool
+ *   itself succeeded or failed - only that the response cannot be delivered.
+ *
+ * - **request-pipeline-failure**: The request was rejected before reaching the
+ *   upstream server due to a required request processing step failure (such as
+ *   authentication, authorization, validation, or rate limiting plugin failure).
+ *
+ * This typically indicates a problem with a plugin or an external system
+ * that a plugin depends on (e.g., audit service, authentication provider).
+ * Retrying is unlikely to help as this usually indicates a configuration
+ * or dependency problem rather than a transient failure.
+ *
+ * @example
+ * ```typescript
+ * import { McpdClient, PipelineError } from '@mozilla-ai/mcpd';
+ *
+ * const client = new McpdClient({ apiEndpoint: 'http://localhost:8090' });
+ *
+ * try {
+ *   const result = await client.servers.time.tools.get_current_time();
+ * } catch (error) {
+ *   if (error instanceof PipelineError) {
+ *     console.log(`Pipeline failure: ${error.message}`);
+ *     console.log(`Flow: ${error.pipelineFlow}`);
+ *
+ *     if (error.pipelineFlow === 'response') {
+ *       console.log('Tool was called but results cannot be delivered');
+ *     } else {
+ *       console.log('Request was rejected by pipeline');
+ *       console.log('Check authentication, authorization, or rate limiting');
+ *     }
+ *   }
+ * }
+ * ```
+ *
+ * @remarks
+ * This exception indicates a problem with a plugin or its dependencies, not
+ * with your request or the tool itself.
+ */
+export class PipelineError extends McpdError {
+  public readonly serverName: string | undefined;
+  public readonly operation: string | undefined;
+  public readonly pipelineFlow: PipelineFlow | undefined;
+
+  constructor(
+    message: string,
+    serverName?: string,
+    operation?: string,
+    pipelineFlow?: PipelineFlow,
+    cause?: Error,
+  ) {
+    super(message, cause);
+    this.name = "PipelineError";
+    this.serverName = serverName;
+    this.operation = operation;
+    this.pipelineFlow = pipelineFlow;
+    Error.captureStackTrace(this, this.constructor);
+  }
+}

src/index.ts

@@ -22,12 +22,16 @@ export {
   McpdError,
   AuthenticationError,
   ConnectionError,
+  PipelineError,
+  PIPELINE_FLOW_REQUEST,
+  PIPELINE_FLOW_RESPONSE,
   ServerNotFoundError,
   ServerUnhealthyError,
   TimeoutError,
   ToolExecutionError,
   ToolNotFoundError,
   ValidationError,
+  type PipelineFlow,
 } from "./errors";
 
 // Export type definitions

tests/unit/client.test.ts

@@ -1,6 +1,11 @@
 import { describe, it, expect, beforeEach, afterEach, vi } from "vitest";
 import { McpdClient } from "../../src/client";
-import { ConnectionError, AuthenticationError } from "../../src/errors";
+import {
+  ConnectionError,
+  AuthenticationError,
+  McpdError,
+  PipelineError,
+} from "../../src/errors";
 import { HealthStatusHelpers } from "../../src/types";
 import { createFetchMock } from "./utils/mockApi";
 import { API_PATHS } from "../../src/apiPaths";
@@ -1286,4 +1291,147 @@ describe("McpdClient", () => {
       );
     });
   });
+
+  describe("PipelineError handling", () => {
+    it("should throw PipelineError for response pipeline failure", async () => {
+      mockFetch.mockResolvedValueOnce({
+        ok: false,
+        status: 500,
+        statusText: "Internal Server Error",
+        headers: new Headers({
+          "Mcpd-Error-Type": "response-pipeline-failure",
+        }),
+        text: async () => "Response processing failed\n",
+      });
+
+      const error = await client.listServers().catch((e: unknown) => e);
+
+      expect(error).toBeInstanceOf(PipelineError);
+      const pipelineError = error as PipelineError;
+      expect(pipelineError.pipelineFlow).toBe("response");
+      expect(pipelineError.message).toContain("Response processing failed");
+    });
+
+    it("should throw PipelineError for request pipeline failure", async () => {
+      mockFetch.mockResolvedValueOnce({
+        ok: false,
+        status: 500,
+        statusText: "Internal Server Error",
+        headers: new Headers({
+          "Mcpd-Error-Type": "request-pipeline-failure",
+        }),
+        text: async () => "Request processing failed\n",
+      });
+
+      const error = await client.listServers().catch((e: unknown) => e);
+
+      expect(error).toBeInstanceOf(PipelineError);
+      const pipelineError = error as PipelineError;
+      expect(pipelineError.pipelineFlow).toBe("request");
+      expect(pipelineError.message).toContain("Request processing failed");
+    });
+
+    it("should throw McpdError for 500 without Mcpd-Error-Type header", async () => {
+      mockFetch.mockResolvedValueOnce({
+        ok: false,
+        status: 500,
+        statusText: "Internal Server Error",
+        headers: new Headers(),
+        text: async () => "Internal Server Error",
+      });
+
+      const error = await client.listServers().catch((e: unknown) => e);
+
+      expect(error).not.toBeInstanceOf(PipelineError);
+      expect(error).toBeInstanceOf(McpdError);
+    });
+
+    it("should handle case-insensitive header value", async () => {
+      mockFetch.mockResolvedValueOnce({
+        ok: false,
+        status: 500,
+        statusText: "Internal Server Error",
+        headers: new Headers({
+          "Mcpd-Error-Type": "RESPONSE-PIPELINE-FAILURE",
+        }),
+        text: async () => "Response processing failed\n",
+      });
+
+      const error = await client.listServers().catch((e: unknown) => e);
+
+      expect(error).toBeInstanceOf(PipelineError);
+      const pipelineError = error as PipelineError;
+      expect(pipelineError.pipelineFlow).toBe("response");
+    });
+
+    it("should enrich PipelineError with server and tool context in performCall", async () => {
+      // First mock: health check for ensureServerHealthy.
+      mockFetch.mockResolvedValueOnce({
+        ok: true,
+        json: async () => ({
+          name: "time",
+          status: "ok",
+          latency: "2ms",
+          lastChecked: "2025-10-07T15:00:00Z",
+          lastSuccessful: "2025-10-07T15:00:00Z",
+        }),
+      });
+
+      // Second mock: tools list for getTools (to verify tool exists).
+      mockFetch.mockResolvedValueOnce({
+        ok: true,
+        json: async () => ({
+          tools: [
+            {
+              name: "get_current_time",
+              description: "Get current time",
+              inputSchema: {
+                type: "object",
+                properties: { timezone: { type: "string" } },
+              },
+            },
+          ],
+        }),
+      });
+
+      // Third mock: tool call returns pipeline error.
+      mockFetch.mockResolvedValueOnce({
+        ok: false,
+        status: 500,
+        statusText: "Internal Server Error",
+        headers: new Headers({
+          "Mcpd-Error-Type": "response-pipeline-failure",
+        }),
+        text: async () => "Response processing failed\n",
+      });
+
+      const error = await client.servers.time!.tools.get_current_time!({
+        timezone: "UTC",
+      }).catch((e: unknown) => e);
+
+      expect(error).toBeInstanceOf(PipelineError);
+      const pipelineError = error as PipelineError;
+      expect(pipelineError.pipelineFlow).toBe("response");
+      expect(pipelineError.serverName).toBe("time");
+      expect(pipelineError.operation).toBe("time.get_current_time");
+      expect(pipelineError.message).toContain("Response processing failed");
+    });
+
+    it("should not throw PipelineError for other 500 errors", async () => {
+      mockFetch.mockResolvedValueOnce({
+        ok: false,
+        status: 500,
+        statusText: "Internal Server Error",
+        headers: new Headers({
+          "Mcpd-Error-Type": "some-other-error-type",
+        }),
+        text: async () => "Some other error",
+      });
+
+      const error = await client.listServers().catch((e: unknown) => e);
+
+      expect(error).not.toBeInstanceOf(PipelineError);
+      expect(error).toBeInstanceOf(McpdError);
+    });
+  });
 });