Developers · Patterns

Integration Patterns

Concrete end-to-end scenarios developers actually build, which endpoints to use, in what order, with enough code to copy-paste.

Back to Developer Hub

Pattern 1, Event bridge from your SaaS to Pulse

Your SaaS fires webhooks (billing, e-commerce, scheduling, internal). Pulse generates a public URL + HMAC secret per source; each verified payload publishes to a topic that wakes up your pipeline.

python

1import requests, hmac, hashlib, json, time
2
3PULSE_URL = "https://pulse.yourdomain.com"
4SECRET = "your-webhook-secret"
5
6def publish_to_pulse(source_id, key, payload_dict):
7    body = json.dumps(payload_dict).encode()
8    signature = hmac.new(SECRET.encode(), body, hashlib.sha256).hexdigest()
9    timestamp = str(int(time.time()))
10    r = requests.post(
11        f"{PULSE_URL}/api/webhooks/{source_id}",
12        headers={
13            "Content-Type": "application/json",
14            "X-Pulse-Signature": signature,
15            "X-Pulse-Timestamp": timestamp,
16            "X-Pulse-Event-Key": key,
17        },
18        data=body,
19        timeout=10,
20    )
21    r.raise_for_status()
22
23publish_to_pulse("billing-ingress", "evt-abc", {"amount": 9900, "customer": "cus_123"})

Pulse verifies the HMAC before publishing the event, rejecting forgeries.

Pattern 2, GitOps for agent deployments

Pipeline definitions live in Git. PR merges apply to staging; a promotion workflow applies to prod. Same apply-script per environment.

yaml

1# .github/workflows/pulse-sync.yaml
2on:
3  push:
4    branches: [main]
5    paths: ['infra/pulse-staging/**']
6
7jobs:
8  sync-staging:
9    runs-on: ubuntu-latest
10    steps:
11      - uses: actions/checkout@v4
12      - name: Apply
13        env:
14          PULSE_URL:  ${{ vars.PULSE_STAGING_URL }}
15          PULSE_USER: ${{ secrets.PULSE_STAGING_USER }}
16          PULSE_PASS: ${{ secrets.PULSE_STAGING_PASS }}
17        run: |
18          LOGIN=$(curl -sf -X POST "$PULSE_URL/api/auth/login" \
19            -H "Content-Type: application/json" \
20            -d "{\"username\":\"$PULSE_USER\",\"password\":\"$PULSE_PASS\"}")
21          export PULSE_TOKEN=$(echo $LOGIN | jq -r '.accessToken')
22          export CSRF=$(echo $LOGIN | jq -r '.csrfToken')
23          cd infra/pulse-staging
24          ../../scripts/apply-agents.sh
25          ../../scripts/apply-pipelines.sh

Pattern 3, Real-time ETL to your data warehouse

Mirror every Pulse decision into your warehouse for BI. Subscribe over WebSocket; insert as events arrive. Use REST polling on /api/pulse/events?since=<cursor> as a backfill companion for at-least-once semantics.

python

1# etl.py, long-running service, subscribes and inserts
2import asyncio, websockets, json, psycopg2, os
3
4PULSE_WS = os.environ["PULSE_WS_URL"]       # ws://pulse:9091/ws
5TOKEN    = os.environ["PULSE_TOKEN"]
6DB_URL   = os.environ["DATABASE_URL"]
7
8async def run():
9    conn = psycopg2.connect(DB_URL)
10    cur = conn.cursor()
11    async with websockets.connect(PULSE_WS) as ws:
12        await ws.send(json.dumps({"type": "auth", "token": TOKEN}))
13        async for raw in ws:
14            msg = json.loads(raw)
15            if msg["type"] == "events":
16                data = msg["data"]
17                cur.execute(
18                    "INSERT INTO pulse_events (id, topic, ts, payload) VALUES (%s, %s, %s, %s::jsonb)",
19                    (data["id"], data["topic"], data["timestamp"], json.dumps(data["payload"])),
20                )
21                conn.commit()
22
23asyncio.run(run())

Pattern 4, Use a Pulse agent as an external tool

You've built a Pulse pipeline that does something useful. You want a desktop assistant or another agent host to call it.

Fastest path, native tool endpoint (no bridge)

Pulse exposes a spec-compliant tool-calling endpoint at POST /tools. Point your host directly at it and every Pulse tool shows up.

json

1{
2  "toolServers": {
3    "pulse": {
4      "type": "http",
5      "url": "http://localhost:9090/tools",
6      "headers": {
7        "Authorization": "Bearer <your-pulse-access-token>"
8      }
9    }
10  }
11}

Alternative, stdio bridge (single named tool)

Useful when your host only speaks stdio, or when you want to expose a specific pipeline as a single named tool with synthesised semantics.

python

1# ~/.pulse-bridge/bridge.py
2import asyncio, os, requests, json, uuid, time
3from tool_server import Server
4import tool_server.stdio
5
6PULSE_URL = os.environ["PULSE_URL"]
7PULSE_TOKEN = os.environ["PULSE_TOKEN"]
8
9srv = Server("pulse-bridge")
10
11@srv.tool()
12async def summarize_inbox(since_iso: str) -> str:
13    """Use the Pulse email-summariser agent to summarise recent emails."""
14    key = f"host-{uuid.uuid4()}"
15    publish = requests.post(f"{PULSE_URL}/api/pulse/events",
16        headers={"Authorization": f"Bearer {PULSE_TOKEN}"},
17        json={"topic": "inbox.summarise.request",
18              "key": key,
19              "value": json.dumps({"since": since_iso})}).json()
20    started = publish["timestamp"]
21    for _ in range(45):
22        time.sleep(1)
23        r = requests.get(f"{PULSE_URL}/api/pulse/events",
24            headers={"Authorization": f"Bearer {PULSE_TOKEN}"},
25            params={"topic": "inbox.summarise.result", "since": started}).json()
26        for e in r.get("events", []):
27            if e.get("key") == key:
28                return e["value"]
29    raise RuntimeError("Timeout")
30
31if __name__ == "__main__":
32    asyncio.run(tool_server.stdio.stdio_server(srv))

Pattern 5, Programmatic agent creation from your wizard

You're building a product on top of Pulse. Users configure their workflow in YOUR UI; you provision the matching Pulse agents behind the scenes, create, schedule, start, update, delete.

typescript

1// backend.ts, your product's API
2import { PulseClient } from './pulse-client';
3
4export async function provisionUserWorkflow(userConfig: UserWorkflow) {
5  const pulse = new PulseClient(process.env.PULSE_URL!);
6  await pulse.login(process.env.PULSE_BOT_USER!, process.env.PULSE_BOT_PASS!);
7
8  const agent = await pulse.call('POST', '/api/pulse/agents', {
9    name: `user-${userConfig.userId}-workflow`,
10    engine: 'llm',
11    inputTopic: `user-${userConfig.userId}.inbound`,
12    outputTopic: `user-${userConfig.userId}.outbound`,
13    config: {
14      systemPrompt: buildPrompt(userConfig),
15      externalTools: userConfig.enabledTools,
16    },
17  });
18
19  await pulse.call('POST', '/api/pulse/schedules', {
20    pipelineOrAgentId: agent.id,
21    cron: userConfig.cron,
22    timezone: userConfig.timezone,
23  });
24
25  await pulse.call('POST', `/api/pulse/agents/${agent.id}/start`);
26  return { agentId: agent.id };
27}

Pattern 6, Monitoring Pulse from Grafana

Pulse exposes Prometheus metrics on /metrics. Scrape them, build dashboards, alert on drift and stale backups.

yaml

1# prometheus.yaml
2scrape_configs:
3  - job_name: 'pulse'
4    scrape_interval: 30s
5    static_configs:
6      - targets: ['pulse.internal:9090']
7    metrics_path: /metrics

pulse_agents_running

Live agents over time.

pulse_agents_drifted

Watchdog couldn't auto-recover.

pulse_events_total

Cumulative event count per topic.

pulse_backup_age_seconds

Hours since last backup; alert > 48h.

Alertmanager rules

yaml

1groups:
2  - name: pulse
3    rules:
4      - alert: PulseAgentsDrifted
5        expr: pulse_agents_drifted > 0
6        for: 10m
7        annotations:
8          summary: "Pulse has {{ $value }} drifted agents, watchdog didn't auto-recover"
9      - alert: PulseBackupStale
10        expr: pulse_backup_age_seconds > 48 * 3600
11        for: 1h

Pattern 7, Bulk-import historical events

Migrating from another system? Backfill Pulse with historical events. Set a backfill: true header so agents can filter, prevents real side effects (emails, notifications) during import.

python

1import requests, json
2
3PULSE = "http://localhost:9090"
4TOKEN = "..."
5
6with open('historical-events.jsonl') as f:
7    for line in f:
8        ev = json.loads(line)
9        requests.post(f"{PULSE}/api/pulse/events",
10            headers={"Authorization": f"Bearer {TOKEN}"},
11            json={
12                "topic": ev["topic"],
13                "key": ev["key"],
14                "value": json.dumps(ev["payload"]),
15                "headers": {
16                    "backfill": "true",
17                    "originalTimestamp": ev["timestamp"],
18                },
19            })

Rate limit: ~300 events/sec on a single-node install. For larger imports, publish in chunks with backoff.

Pattern 8, Feature-flag pipeline variants

A/B test a new prompt across 10% of your traffic. Route deterministically with a rule-based agent, then diff outputs between live and candidate.

yaml

1# agents/email-triager-router.yaml, routes 10% to candidate, 90% to live
2name: email-triager-router
3engine: rule-based
4inputTopic: email.ingress
5outputTopic: email.ingress.10pct
6config:
7  rules:
8    - condition: "hash(messageId) % 10 == 0"
9      action: emit
10      target: email.ingress.10pct
11    - condition: always
12      action: emit
13      target: email.ingress.90pct

Or use Pulse's native shadow-deploy feature, which does this without rule-based routing.

Pattern 9, Multi-tenant Pulse (MSP scenario)

Option 1, One install per client

Simplest. Each client gets their own data directory, domain, license. Run N processes behind a reverse proxy; route by subdomain. Works out of the box.

Option 2, One install, multi-org

Enterprise feature. Multiple orgs share one Pulse binary; per-org data isolation; per-org users. Requires Enterprise licensing.

Pattern 10, Embedding Pulse in your product

White-label the UI, expose your own API to end-users, hold the Pulse admin credentials in your backend. End-users authenticate against YOUR auth, not Pulse's.

Commercial redistribution requires an Enterprise agreement, get in touch before shipping.

Need help with a custom pattern?

Our team can review your architecture and recommend the cleanest integration shape.