How to Integrate with Pipedream

Learn how to use the SnackPrompt AI Engine API to build powerful automations and AI-powered workflows in Pipedream.

Overview

Pipedream is a developer-focused integration platform with native code support. It offers several ways to integrate APIs:

Method	Use Case	Description
HTTP Request Action	No-code	Simple API calls without coding
Node.js Code Step	Full control	Custom JavaScript/TypeScript code
Python Code Step	Data processing	Python for complex transformations
Custom Component	Reusable	Create shareable integration components

Integration Architecture

┌─────────────────────────────────────────────────────────┐
│                      Pipedream                          │
│  ┌─────────────┐    ┌─────────────┐    ┌────────────┐   │
│  │   Trigger   │──▶│ Code/HTTP    │──▶│   Action   │   │
│  │   Source    │    │    Step     │    │            │   │
│  └─────────────┘    └──────┬──────┘    └────────────┘   │
│                            │                            │
│                     ┌──────▼──────┐                     │
│                     │   $export   │                     │
│                     │   (state)   │                     │
│                     └──────┬──────┘                     │
└────────────────────────────┼────────────────────────────┘
                             │
                             ▼
              ┌──────────────────────────────┐
              │  SnackPrompt AI Engine API   │
              │  /v1/kb/search or /v1/kb/chat│
              └──────────────────────────────┘

---

Method 1: Node.js Code Step (Recommended)

Pipedream excels at code-based integrations. Use Node.js for full control.

Step 1: Create a New Workflow

1. Go to pipedream.com

2. Click New > Workflow

3. Choose your trigger (HTTP, Schedule, App event, etc.)

Step 2: Add Node.js Code Step

1. Click + to add a step

2. Select Code > Node.js

3. Add the code below

Search Knowledge Base

import { axios } from "@pipedream/platform";

export default defineComponent({
  props: {
    snackprompt_api_key: {
      type: "string",
      label: "SnackPrompt API Key",
      secret: true,
    },
    tenant_id: {
      type: "string",
      label: "Tenant ID",
    },
  },
  async run({ steps, $ }) {
    const query = steps.trigger.event.body?.query || steps.trigger.event.query;

    const response = await axios($, {
      method: "POST",
      url: "https://api-integrations.snackprompt.com/v1/kb/search",
      headers: {
        "Content-Type": "application/json",
        "x-api-key": this.snackprompt_api_key,
      },
      data: {
        query: query,
        filters: {
          tenant_id: this.tenant_id,
        },
        limit: 5,
      },
    });

    // Format results for easier use
    const results = response.items.map((item, index) => ({
      position: index + 1,
      text: item.payload.original_text,
      score: item.score,
      id: item.payload.snack_item_id,
    }));

    return {
      results,
      context: results.map(r => `[${r.position}] ${r.text}`).join('\n\n'),
      count: results.length,
    };
  },
});

Chat with Knowledge Base

import { axios } from "@pipedream/platform";

export default defineComponent({
  props: {
    snackprompt_api_key: {
      type: "string",
      label: "SnackPrompt API Key",
      secret: true,
    },
    tenant_id: {
      type: "string",
      label: "Tenant ID",
    },
    tag_names: {
      type: "string[]",
      label: "Tag Names",
      optional: true,
      default: [],
    },
  },
  async run({ steps, $ }) {
    const query = steps.trigger.event.body?.message ||
                  steps.trigger.event.body?.query ||
                  steps.trigger.event.query;

    const response = await axios($, {
      method: "POST",
      url: "https://api-integrations.snackprompt.com/v1/kb/chat",
      headers: {
        "Content-Type": "application/json",
        "x-api-key": this.snackprompt_api_key,
      },
      data: {
        query: query,
        filters: {
          tenant_id: this.tenant_id,
          tag_names: this.tag_names,
        },
      },
    });

    return {
      answer: response.answer,
      sources: response.sources,
      hasAnswer: !!response.answer,
    };
  },
});

---

Method 2: HTTP Request Action (No-Code)

For simple integrations without custom code.

Step 1: Add HTTP Request Step

1. Click + to add a step

2. Search for HTTP / Webhook

3. Select Send any HTTP Request

Step 2: Configure Request

Field	Value
Method	POST
URL	https://api-integrations.snackprompt.com/v1/kb/search

Headers:

Key	Value
Content-Type	application/json
x-api-key	{{steps.trigger.event.headers["x-api-key"]}} or configure in Auth

Body:

{
  "query": "{{steps.trigger.event.body.query}}",
  "filters": {
    "tenant_id": "YOUR_TENANT_ID"
  },
  "limit": 5
}

Step 3: Use Results

Reference in subsequent steps:

• {{steps.http_request.$return_value.items}}

• {{steps.http_request.$return_value.items[0].payload.original_text}}

---

Method 3: Python Code Step

For data scientists or Python-preferred developers.

Search with Python

import requests

def handler(pd: "pipedream"):
    query = pd.steps["trigger"]["event"]["body"].get("query", "")

    response = requests.post(
        "https://api-integrations.snackprompt.com/v1/kb/search",
        headers={
            "Content-Type": "application/json",
            "x-api-key": pd.inputs["snackprompt_api_key"],
        },
        json={
            "query": query,
            "filters": {
                "tenant_id": pd.inputs["tenant_id"],
            },
            "limit": 5,
        },
    )

    data = response.json()

    # Format results
    results = []
    for i, item in enumerate(data.get("items", [])):
        results.append({
            "position": i + 1,
            "text": item["payload"]["original_text"],
            "score": item["score"],
            "id": item["payload"]["snack_item_id"],
        })

    # Build context string
    context = "\n\n".join([
        f"[{r['position']}] {r['text']}"
        for r in results
    ])

    return {
        "results": results,
        "context": context,
        "count": len(results),
    }

---

Method 4: Custom Component (Reusable)

Create a reusable component for your organization.

Create Component

// snackprompt-search.mjs
import { axios } from "@pipedream/platform";

export default {
  name: "SnackPrompt Search",
  description: "Search the SnackPrompt knowledge base",
  key: "snackprompt-search",
  version: "1.0.0",
  type: "action",
  props: {
    snackprompt: {
      type: "app",
      app: "snackprompt",
    },
    query: {
      type: "string",
      label: "Search Query",
      description: "The query to search for",
    },
    tenant_id: {
      type: "string",
      label: "Tenant ID",
      description: "Your SnackPrompt tenant ID",
    },
    tag_names: {
      type: "string[]",
      label: "Tags",
      description: "Filter by specific tags",
      optional: true,
    },
    limit: {
      type: "integer",
      label: "Limit",
      description: "Maximum number of results",
      default: 5,
      optional: true,
    },
  },
  async run({ $ }) {
    const response = await axios($, {
      method: "POST",
      url: "https://api-integrations.snackprompt.com/v1/kb/search",
      headers: {
        "Content-Type": "application/json",
        "x-api-key": `${this.snackprompt.$auth.api_key}`,
      },
      data: {
        query: this.query,
        filters: {
          tenant_id: this.tenant_id,
          tag_names: this.tag_names || [],
        },
        limit: this.limit,
      },
    });

    $.export("$summary", `Found ${response.items?.length || 0} results`);

    return response;
  },
};

App Definition

// snackprompt.app.mjs
export default {
  type: "app",
  app: "snackprompt",
  propDefinitions: {
    tenant_id: {
      type: "string",
      label: "Tenant ID",
      description: "Your SnackPrompt tenant ID",
    },
  },
  methods: {
    _apiKey() {
      return this.$auth.api_key;
    },
    _baseUrl() {
      return "https://api-integrations.snackprompt.com/v1/kb";
    },
    async _makeRequest(opts = {}) {
      const { $ = this, path, ...otherOpts } = opts;
      return axios($, {
        ...otherOpts,
        url: `${this._baseUrl()}${path}`,
        headers: {
          "Content-Type": "application/json",
          "x-api-key": this._apiKey(),
        },
      });
    },
    async search(args = {}) {
      return this._makeRequest({
        method: "POST",
        path: "/search",
        ...args,
      });
    },
    async chat(args = {}) {
      return this._makeRequest({
        method: "POST",
        path: "/chat",
        ...args,
      });
    },
  },
};

---

Practical Use Cases

1. Webhook-Based Chatbot

// Trigger: HTTP Webhook
// Receives POST with { "query": "user question" }

export default defineComponent({
  async run({ steps, $ }) {
    const query = steps.trigger.event.body.query;

    // Search knowledge base
    const response = await axios($, {
      method: "POST",
      url: "https://api-integrations.snackprompt.com/v1/kb/chat",
      headers: {
        "Content-Type": "application/json",
        "x-api-key": process.env.SNACKPROMPT_API_KEY,
      },
      data: {
        query,
        filters: { tenant_id: process.env.SNACKPROMPT_TENANT_ID },
      },
    });

    // Return response to webhook caller
    await $.respond({
      status: 200,
      headers: { "Content-Type": "application/json" },
      body: {
        answer: response.answer,
        sources: response.sources,
      },
    });
  },
});

2. Slack Bot

// Trigger: Slack - New Message in Channel

export default defineComponent({
  async run({ steps, $ }) {
    const message = steps.trigger.event.text;
    const channel = steps.trigger.event.channel;

    // Search knowledge base
    const searchResult = await axios($, {
      method: "POST",
      url: "https://api-integrations.snackprompt.com/v1/kb/chat",
      headers: {
        "Content-Type": "application/json",
        "x-api-key": process.env.SNACKPROMPT_API_KEY,
      },
      data: {
        query: message,
        filters: { tenant_id: process.env.SNACKPROMPT_TENANT_ID },
      },
    });

    // Post reply to Slack
    await axios($, {
      method: "POST",
      url: "https://slack.com/api/chat.postMessage",
      headers: {
        Authorization: `Bearer ${process.env.SLACK_BOT_TOKEN}`,
        "Content-Type": "application/json",
      },
      data: {
        channel,
        text: searchResult.answer || "I couldn't find relevant information.",
        thread_ts: steps.trigger.event.ts,
      },
    });

    return { success: true };
  },
});

3. Email Auto-Responder

// Trigger: Gmail - New Email

export default defineComponent({
  async run({ steps, $ }) {
    const email = steps.trigger.event;
    const question = `${email.subject}\n\n${email.snippet}`;

    // Get answer from knowledge base
    const response = await axios($, {
      method: "POST",
      url: "https://api-integrations.snackprompt.com/v1/kb/chat",
      headers: {
        "Content-Type": "application/json",
        "x-api-key": process.env.SNACKPROMPT_API_KEY,
      },
      data: {
        query: question,
        filters: {
          tenant_id: process.env.SNACKPROMPT_TENANT_ID,
          tag_names: ["Support"],
        },
      },
    });

    if (response.sources?.length > 0) {
      // Has good answer - auto-reply
      return {
        shouldReply: true,
        answer: response.answer,
        to: email.from,
        subject: `Re: ${email.subject}`,
      };
    } else {
      // No answer - flag for human review
      return {
        shouldReply: false,
        reason: "No relevant information found",
      };
    }
  },
});

4. RAG with OpenAI

import { axios } from "@pipedream/platform";
import OpenAI from "openai";

export default defineComponent({
  props: {
    openai_api_key: { type: "string", secret: true },
    snackprompt_api_key: { type: "string", secret: true },
    tenant_id: { type: "string" },
  },
  async run({ steps, $ }) {
    const query = steps.trigger.event.body.query;

    // 1. Search knowledge base
    const searchResponse = await axios($, {
      method: "POST",
      url: "https://api-integrations.snackprompt.com/v1/kb/search",
      headers: {
        "Content-Type": "application/json",
        "x-api-key": this.snackprompt_api_key,
      },
      data: {
        query,
        filters: { tenant_id: this.tenant_id },
        limit: 5,
      },
    });

    // 2. Format context
    const context = searchResponse.items
      .map((item, i) => `[${i + 1}] ${item.payload.original_text}`)
      .join("\n\n");

    // 3. Generate response with OpenAI
    const openai = new OpenAI({ apiKey: this.openai_api_key });

    const completion = await openai.chat.completions.create({
      model: "gpt-4",
      messages: [
        {
          role: "system",
          content: `Answer based ONLY on this context:\n\n${context}\n\nIf the context doesn't contain relevant information, say so.`,
        },
        { role: "user", content: query },
      ],
    });

    return {
      answer: completion.choices[0].message.content,
      sources: searchResponse.items.map(item => ({
        id: item.payload.snack_item_id,
        score: item.score,
      })),
    };
  },
});

5. Scheduled Knowledge Sync

// Trigger: Schedule - Every day at 9 AM

export default defineComponent({
  async run({ steps, $ }) {
    const topics = ["product updates", "policy changes", "new features"];
    const results = [];

    for (const topic of topics) {
      const response = await axios($, {
        method: "POST",
        url: "https://api-integrations.snackprompt.com/v1/kb/search",
        headers: {
          "Content-Type": "application/json",
          "x-api-key": process.env.SNACKPROMPT_API_KEY,
        },
        data: {
          query: topic,
          filters: { tenant_id: process.env.SNACKPROMPT_TENANT_ID },
          limit: 3,
        },
      });

      if (response.items?.length > 0) {
        results.push({
          topic,
          items: response.items.map(i => i.payload.original_text),
        });
      }
    }

    return { digest: results, date: new Date().toISOString() };
  },
});

---

Environment Variables

Store sensitive data in Pipedream environment variables:

1. Go to Settings > Environment Variables

2. Add:

   • SNACKPROMPT_API_KEY

   • SNACKPROMPT_TENANT_ID

Access in code:

process.env.SNACKPROMPT_API_KEY
process.env.SNACKPROMPT_TENANT_ID

---

Configuration Tips

1. Use Props for Configurability

props: {
  limit: {
    type: "integer",
    label: "Results Limit",
    default: 5,
    min: 1,
    max: 20,
  },
  tags: {
    type: "string[]",
    label: "Filter Tags",
    optional: true,
  },
}

2. Error Handling

try {
  const response = await axios($, { ... });
  return response;
} catch (error) {
  if (error.response?.status === 429) {
    // Rate limited - implement backoff
    await new Promise(r => setTimeout(r, 1000));
    // Retry...
  }
  throw new Error(`API Error: ${error.message}`);
}

3. Data Validation

const query = steps.trigger.event.body?.query;

if (!query || typeof query !== 'string' || query.trim() === '') {
  throw new Error('Query is required and must be a non-empty string');
}

4. Response Caching

import { axios } from "@pipedream/platform";

export default defineComponent({
  async run({ steps, $ }) {
    const query = steps.trigger.event.body.query;
    const cacheKey = `search:${query.toLowerCase().trim()}`;

    // Check cache (using Pipedream's data store)
    const cached = await $.service.db.get(cacheKey);
    if (cached && Date.now() - cached.timestamp < 5 * 60 * 1000) {
      return cached.data;
    }

    // Fetch fresh data
    const response = await axios($, { ... });

    // Cache result
    await $.service.db.set(cacheKey, {
      data: response,
      timestamp: Date.now(),
    });

    return response;
  },
});

5. Parallel Requests

const [salesResults, supportResults] = await Promise.all([
  axios($, {
    method: "POST",
    url: "https://api-integrations.snackprompt.com/v1/kb/search",
    headers: { ... },
    data: {
      query,
      filters: { tenant_id, tag_names: ["Sales"] },
    },
  }),
  axios($, {
    method: "POST",
    url: "https://api-integrations.snackprompt.com/v1/kb/search",
    headers: { ... },
    data: {
      query,
      filters: { tenant_id, tag_names: ["Support"] },
    },
  }),
]);

return {
  sales: salesResults.items,
  support: supportResults.items,
};

---

Troubleshooting

Error: "tenant_id is required"

Ensure tenant_id is inside the filters object:

// ❌ Wrong
{ query: "...", tenant_id: "..." }

// ✅ Correct
{ query: "...", filters: { tenant_id: "..." } }

Error: "Request failed with status 401"

1. Check API key is correct

2. Verify API key is not expired

3. Ensure header name is exactly x-api-key

Empty Results

1. Verify tenant_id is correct

2. Remove tag_names filter to search all content

3. Check query string is not empty

4. Test API directly with curl

Workflow Times Out

1. Reduce limit parameter

2. Add timeout to axios request:

   await axios($, { timeout: 30000, ... })

3. Consider async/webhook pattern for long operations

---

Related

• Endpoints Reference

• Available Filters

• Error Hand

External Resources

• Pipedream Documentation

• Pipedream Node.js Code Steps

• Pipedream Python Code Steps

• Building Custom Components

• Pipedream Data Stores