mcpengine/servers/pipedrive/src/client.ts

// ============================================================================
// Pipedrive API v1 HTTP Client
// ============================================================================

const BASE_URL = "https://api.pipedrive.com/v1";

export class PipedriveClient {
  private apiToken: string;
  private baseUrl: string;
  private rateLimitRemaining = 100;
  private rateLimitReset = 0;

  constructor(apiToken?: string, baseUrl?: string) {
    this.apiToken = apiToken || process.env.PIPEDRIVE_API_TOKEN || "";
    this.baseUrl = baseUrl || process.env.PIPEDRIVE_BASE_URL || BASE_URL;

    if (!this.apiToken) {
      throw new Error(
        "Pipedrive API token is required. Set PIPEDRIVE_API_TOKEN environment variable."
      );
    }
  }

  private buildUrl(path: string, query?: Record<string, unknown>): string {
    const url = new URL(path, this.baseUrl);
    // Add API token to all requests
    url.searchParams.set("api_token", this.apiToken);
    
    if (query) {
      for (const [key, value] of Object.entries(query)) {
        if (value !== undefined && value !== null && value !== "") {
          url.searchParams.set(key, String(value));
        }
      }
    }
    return url.toString();
  }

  private get headers(): Record<string, string> {
    return {
      "Content-Type": "application/json",
      Accept: "application/json",
      "User-Agent": "pipedrive-mcp/1.0.0",
    };
  }

  private async checkRateLimit(): Promise<void> {
    if (this.rateLimitRemaining <= 1 && Date.now() < this.rateLimitReset) {
      const waitMs = this.rateLimitReset - Date.now();
      if (waitMs > 0 && waitMs < 60000) {
        await new Promise((resolve) => setTimeout(resolve, waitMs));
      }
    }
  }

  private updateRateLimit(response: Response): void {
    const remaining = response.headers.get("x-ratelimit-remaining");
    const reset = response.headers.get("x-ratelimit-reset");
    if (remaining) this.rateLimitRemaining = parseInt(remaining, 10);
    if (reset) this.rateLimitReset = parseInt(reset, 10) * 1000;
  }

  async get<T = unknown>(
    path: string,
    query?: Record<string, unknown>
  ): Promise<T> {
    await this.checkRateLimit();
    const url = this.buildUrl(path, query);
    const response = await fetch(url, {
      method: "GET",
      headers: this.headers,
    });
    this.updateRateLimit(response);
    return this.handleResponse<T>(response);
  }

  async post<T = unknown>(
    path: string,
    body?: unknown,
    query?: Record<string, unknown>
  ): Promise<T> {
    await this.checkRateLimit();
    const url = this.buildUrl(path, query);
    const options: RequestInit = {
      method: "POST",
      headers: this.headers,
    };
    if (body !== undefined) {
      options.body = JSON.stringify(body);
    }
    const response = await fetch(url, options);
    this.updateRateLimit(response);
    return this.handleResponse<T>(response);
  }

  async put<T = unknown>(
    path: string,
    body?: unknown,
    query?: Record<string, unknown>
  ): Promise<T> {
    await this.checkRateLimit();
    const url = this.buildUrl(path, query);
    const options: RequestInit = {
      method: "PUT",
      headers: this.headers,
    };
    if (body !== undefined) {
      options.body = JSON.stringify(body);
    }
    const response = await fetch(url, options);
    this.updateRateLimit(response);
    return this.handleResponse<T>(response);
  }

  async delete<T = unknown>(
    path: string,
    query?: Record<string, unknown>
  ): Promise<T> {
    await this.checkRateLimit();
    const url = this.buildUrl(path, query);
    const response = await fetch(url, {
      method: "DELETE",
      headers: this.headers,
    });
    this.updateRateLimit(response);
    return this.handleResponse<T>(response);
  }

  async postFormData<T = unknown>(
    path: string,
    formData: FormData,
    query?: Record<string, unknown>
  ): Promise<T> {
    await this.checkRateLimit();
    const url = this.buildUrl(path, query);
    const headers: Record<string, string> = {
      Accept: "application/json",
      "User-Agent": "pipedrive-mcp/1.0.0",
    };
    const response = await fetch(url, {
      method: "POST",
      headers,
      body: formData,
    });
    this.updateRateLimit(response);
    return this.handleResponse<T>(response);
  }

  private async handleResponse<T>(response: Response): Promise<T> {
    if (!response.ok) {
      let errorBody: string;
      try {
        const json = await response.json() as any;
        errorBody = json.error || json.message || JSON.stringify(json);
      } catch {
        errorBody = await response.text().catch(() => "Unable to read error body");
      }
      throw new ApiError(
        `Pipedrive API error ${response.status}: ${errorBody}`,
        response.status,
        errorBody
      );
    }

    try {
      const json = await response.json() as any;
      // Pipedrive wraps responses in { success: true, data: ... }
      if (json.success === false) {
        throw new ApiError(
          `Pipedrive API error: ${json.error || "Unknown error"}`,
          response.status,
          JSON.stringify(json)
        );
      }
      return json as T;
    } catch (e) {
      if (e instanceof ApiError) throw e;
      // Some endpoints return 204 No Content
      if (response.status === 204) {
        return { success: true } as T;
      }
      throw new ApiError(
        `Failed to parse response: ${(e as Error).message}`,
        response.status,
        ""
      );
    }
  }

  /**
   * Handle paginated responses. Pipedrive uses `start` and `limit` parameters.
   */
  async *paginate<T>(
    path: string,
    query: Record<string, unknown> = {},
    limitPerPage = 100
  ): AsyncGenerator<T[], void, unknown> {
    let start = 0;
    let hasMore = true;

    while (hasMore) {
      const result = await this.get<{
        success: boolean;
        data: T[];
        additional_data?: {
          pagination?: {
            start: number;
            limit: number;
            more_items_in_collection: boolean;
            next_start?: number;
          };
        };
      }>(path, { ...query, start, limit: limitPerPage });

      if (result.data && result.data.length > 0) {
        yield result.data;
      }

      const pagination = result.additional_data?.pagination;
      hasMore = pagination?.more_items_in_collection ?? false;
      start = pagination?.next_start ?? start + limitPerPage;

      if (!hasMore || !result.data || result.data.length === 0) {
        break;
      }
    }
  }
}

export class ApiError extends Error {
  constructor(
    message: string,
    public status: number,
    public body: string
  ) {
    super(message);
    this.name = "ApiError";
  }
}

/**
 * Helper to wrap errors into MCP-compatible tool results
 */
export function err(error: unknown): {
  content: Array<{ type: "text"; text: string }>;
  isError: true;
} {
  const msg =
    error instanceof ApiError
      ? `${error.message}\nStatus: ${error.status}\nBody: ${error.body}`
      : error instanceof Error
        ? error.message
        : String(error);

  return {
    content: [{ type: "text", text: `Error: ${msg}` }],
    isError: true,
  };
}