---
title: "Pipedream"
description: "通过 TwexAPI 构建 Pipedream 工作流自动化，实现推文搜索、资料查询、热门推文、HTTP 触发与经批准发帖。"
---

用 TwexAPI 构建 Pipedream Twitter 集成，用于搜索、资料、热门推文与经批准发帖。TwexAPI 为每个工作流提供单一 REST API 与单一 API 密钥。

AI Agent handoff 用 webhook 触发， recurring 搜索用 schedule 触发，需要自定义规范化时用内联 code 步骤。

## 选择 Pipedream 自动化模式

<CardGroup cols={2}>
  <Card title="HTTP trigger + code step" icon="webhook">
    接收 MCP AI Agent严格 JSON，规范化行并路由到 Slack 或 Sheets。
  </Card>

  <Card title="定时 REST 数据读取" icon="calendar-clock">
    按 cadence 调用 TwexAPI 做推文搜索或趋势报告。
  </Card>

  <Card title="私有组件扩展包" icon="boxes">
    团队需要相同操作时打包重复认证请求。
  </Card>
</CardGroup>

将搜索、webhook handoff、粉丝导出与经批准写操作拆成独立工作流。更小的工作流更易暴露错误与速率限制。

## 前置条件

* [TwexAPI API 密钥](https://twexapi.io/dashboard)
* Pipedream 账号
* 可选 Slack、Google Sheets、Airtable 或数据库连接账号
* 可选：已连接 `https://api.twexapi.io/mcp` 的 MCP capable AI Agent

## Serverless API 集成

每个 TwexAPI 请求从 `https://api.twexapi.io` 开始。通过 `Authorization: Bearer` header 发送 API 密钥。密钥放在 Pipedream 环境变量，勿放在步骤导出或日志。

```bash
export TWEXAPI_API_KEY="YOUR_API_KEY"
```

首次认证检查用 `GET /balance`，验证 API 密钥且不修改 X 账号。

## 共享请求 helper

```javascript
export default defineComponent({
  props: {
    twexapi: {
      type: "string",
      label: "TwexAPI API Key",
      secret: true,
    },
  },
  methods: {
    async twexapiRequest($, { method, path, body, params }) {
      const url = new URL(`https://api.twexapi.io${path}`);
      if (params) {
        Object.entries(params).forEach(([key, value]) => {
          if (value !== undefined && value !== null) {
            url.searchParams.set(key, String(value));
          }
        });
      }

      const response = await fetch(url, {
        method,
        headers: {
          Authorization: `Bearer ${this.twexapi}`,
          "Content-Type": "application/json",
        },
        body: body ? JSON.stringify(body) : undefined,
      });

      if (!response.ok) {
        throw new Error(`TwexAPI request failed with HTTP ${response.status}`);
      }

      return response.json();
    },
  },
});
```

## 入门操作

| 操作 | TwexAPI 路由 |
| --- | --- |
| 搜索推文 | `POST /twitter/advanced_search/page` |
| 获取资料 | `GET /twitter/{screen_name}/about` |
| 搜索用户 | `GET /twitter/search-user/{keyword}/{target_count}` |
| 获取趋势 | `GET /twitter/global-trending/tweets` |
| 列出粉丝 | `POST /v3/twitter/users/followers` |
| 发帖 | `POST /twitter/tweets/create` |

### 搜索推文 code 步骤

```javascript
export default defineComponent({
  async run({ steps, $ }) {
    const response = await fetch("https://api.twexapi.io/twitter/advanced_search/page", {
      method: "POST",
      headers: {
        Authorization: `Bearer ${process.env.TWEXAPI_API_KEY}`,
        "Content-Type": "application/json",
      },
      body: JSON.stringify({
        searchTerms: ["AI agents lang:en"],
        sortBy: "Latest",
        nextCursor: "",
      }),
    });

    if (!response.ok) {
      throw new Error(`TwexAPI request failed with HTTP ${response.status}`);
    }

    return response.json();
  },
});
```

## Webhook 优先流程

1. 在 Pipedream 创建 HTTP 触发器。
2. 让 MCP capable AI Agent采集 TwexAPI 行并返回严格 JSON。
3. 将 JSON POST 到 Pipedream 端点。
4. 在 code 步骤规范化并去重行。

### JavaScript 规范化步骤

```javascript
export default defineComponent({
  async run({ steps }) {
    const payload = steps.trigger.event.body;

    return (payload.tweets || []).map((tweet) => ({
      id: tweet.tweet_id || tweet.id,
      url: tweet.public_url ?? `https://x.com/i/web/status/${tweet.tweet_id || tweet.id}`,
      author: tweet.author_username,
      text: tweet.text || tweet.full_text,
      created_at: tweet.created_at,
      route_used: payload.route_used,
      next_cursor: payload.next_cursor,
    }));
  },
});
```

建议 MCP AI Agent prompt：

```text
Use Twexapi MCP to get quote tweets for tweet ID 1803006263529541838.
Return only JSON with route_used, source_tweet_id, has_more, next_cursor, and tweets.
Each tweet must include tweet_id, author_username, text, created_at, and public_url.
Do not call write endpoints.
```

## 结果 handoff

<CardGroup cols={2}>
  <Card title="推文分页数据" icon="search">
    导出 `tweet_count`、`has_more`、`next_cursor`；返回含 `tweet_id`、`text`、`author_username`、`created_at` 的行。
  </Card>

  <Card title="用户资料数据" icon="user-round">
    导出 `user_id`、`username`、`name`、粉丝数与 verified 字段。
  </Card>

  <Card title="趋势话题批次" icon="trending-up">
    导出国家、主题、内容标签与推文行及工作流元数据。
  </Card>

  <Card title="写入操作 (发推/点赞/关注)" icon="send">
    用 [CLI](/sdks/cli) `--dry-run` 预览。写操作需 cookie 或 `auth_token`。
  </Card>
</CardGroup>

## 错误处理

导出下游行之前按状态路由。

| 状态 | 操作 |
| ------ | ------ |
| `400` | 修正请求字段；切勿原样重试。 |
| `401` | 更换 API 密钥。 |
| `403` | 解决账号访问或额度。 |
| `429` | 退避并保留游标。 |
| `5xx` | 对有界退避重试安全读取。 |

仅当下游处理成功后再持久化页游标。

## 配方

### 搜索推文到 Slack

1. 按报告 cadence Schedule 工作流。
2. 调用 `POST /twitter/advanced_search/page`。
3. 按互动阈值过滤推文。
4. 将选定推文文本、作者与链接发到 Slack。

### AI Agent handoff 到 Google Sheets

1. HTTP 触发器接收 MCP handoff JSON。
2. 规范化推文行。
3. 按 `tweet_id` upsert 到 Sheets。

### 粉丝页到 CRM

1. HTTP 或定时步骤调用 `POST /v3/twitter/users/followers`。
2. 规范化粉丝行。
3. CRM 按 `user_id` upsert。
4. 数据存储保留 `next_cursor`。

## 测试清单

* 确认触发 body 为有效 JSON。
* 以 `tweet_id` 或 `user_id` 为去重键。
* 在 Pipedream 数据存储保存 `next_cursor` 供定时续跑。
* `TWEXAPI_API_KEY` 仅放在环境变量。
* 切勿在步骤输出中记录 Bearer token。

## 下一步

* [Make](/guides/make)
* [Zapier](/guides/zapier)
* [n8n](/guides/n8n)
* [Agent MCP Handoff](/mcp/agent-handoff)
* [API Reference](/api-reference/overview)