AI Workflow Examples

These examples show how AI tools can help you integrate with Emerge more efficiently using the MCP connection.

Example 1: Generate integration code

You ask:

“Generate TypeScript and Python code to create a signed Emerge link and handle the callback”

AI response (with MCP connection): The AI searches the docs, finds the Create Links and Callbacks pages, and generates code that keeps uid server-side (never user input):

import express from 'express';
import crypto from 'crypto';

const app = express();

const CLIENT_ID = process.env.EMERGE_CLIENT_ID;
const SIGNING_SECRET = process.env.EMERGE_SIGNING_SECRET;
const REDIRECT_URI = process.env.EMERGE_REDIRECT_URI || 'https://yourapp.com/emerge/callback';

if (!CLIENT_ID || !SIGNING_SECRET) {
  throw new Error('Missing EMERGE_CLIENT_ID or EMERGE_SIGNING_SECRET');
}

const stateStore = new Map<string, { internalUserId: string; createdAt: number }>();
const uidMap = new Map<string, string>();

function createLinkUrl(internalUserId: string): { url: string; state: string } {
  const timestamp = new Date().toISOString();
  const state = crypto.randomBytes(16).toString('hex');

  const params: Record<string, string> = {
    client_id: CLIENT_ID,
    redirect_uri: REDIRECT_URI,
    state,
    timestamp
  };

  // Use your internal user id or omit uid entirely.
  if (internalUserId) {
    params.uid = internalUserId;
  }

  const signatureBase = Object.keys(params)
    .sort()
    .map(key => `${key}=${params[key]}`)
    .join('&');

  const signature = crypto
    .createHmac('sha256', SIGNING_SECRET)
    .update(signatureBase)
    .digest('hex');

  const finalParams = new URLSearchParams(params);
  finalParams.append('signature', signature);

  return { url: `https://link.emergedata.ai/link/start?${finalParams.toString()}`, state };
}

app.get('/emerge/link', (req, res) => {
  try {
    const internalUserId = String(req.query.user_id || '');
    if (!internalUserId) {
      return res.status(400).send('Missing user_id');
    }

    const { url, state } = createLinkUrl(internalUserId);
    stateStore.set(state, { internalUserId, createdAt: Date.now() });

    return res.redirect(url);
  } catch (err) {
    console.error('Failed to create link', err);
    return res.status(500).send('Server error');
  }
});

app.get('/emerge/callback', async (req, res) => {
  try {
    const { status, state, uid, error_code } = req.query as Record<string, string>;
    const record = stateStore.get(state);

    if (!record) {
      return res.status(400).send('Invalid state');
    }

    stateStore.delete(state);

    if (status === 'success' || status === 'reauthorized') {
      if (!uid) {
        return res.status(400).send('Missing uid');
      }

      uidMap.set(record.internalUserId, uid);
      return res.redirect('/dashboard?connected=true');
    }

    const errorMessage = encodeURIComponent(error_code || 'unknown_error');
    return res.redirect(`/connect?error=${errorMessage}`);
  } catch (err) {
    console.error('Callback error', err);
    return res.status(500).send('Server error');
  }
});

app.listen(3000, () => {
  console.log('Server running on http://localhost:3000');
});

import os
import hmac
import hashlib
import secrets
from datetime import datetime, timezone
from urllib.parse import urlencode

from fastapi import FastAPI, HTTPException
from fastapi.responses import RedirectResponse
import uvicorn

CLIENT_ID = os.getenv('EMERGE_CLIENT_ID')
SIGNING_SECRET = os.getenv('EMERGE_SIGNING_SECRET')
REDIRECT_URI = os.getenv('EMERGE_REDIRECT_URI', 'https://yourapp.com/emerge/callback')

if not CLIENT_ID or not SIGNING_SECRET:
    raise RuntimeError('Missing EMERGE_CLIENT_ID or EMERGE_SIGNING_SECRET')

app = FastAPI()

state_store: dict[str, dict] = {}
uid_map: dict[str, str] = {}


def create_link_url(internal_user_id: str) -> tuple[str, str]:
    timestamp = datetime.now(timezone.utc).isoformat().replace('+00:00', 'Z')
    state = secrets.token_hex(16)

    params = {
        'client_id': CLIENT_ID,
        'redirect_uri': REDIRECT_URI,
        'state': state,
        'timestamp': timestamp
    }

    # Use your internal user id or omit uid entirely.
    if internal_user_id:
        params['uid'] = internal_user_id

    signature_base = '&'.join(f'{k}={v}' for k, v in sorted(params.items()))

    signature = hmac.new(
        SIGNING_SECRET.encode(),
        signature_base.encode(),
        hashlib.sha256
    ).hexdigest()

    params['signature'] = signature
    return f"https://link.emergedata.ai/link/start?{urlencode(params)}", state


@app.get('/emerge/link')
async def emerge_link(user_id: str):
    if not user_id:
        raise HTTPException(status_code=400, detail='Missing user_id')

    url, state = create_link_url(user_id)
    state_store[state] = {
        'internal_user_id': user_id,
        'created_at': datetime.now(timezone.utc).timestamp()
    }

    return RedirectResponse(url)


@app.get('/emerge/callback')
async def emerge_callback(
    status: str,
    state: str,
    uid: str = '',
    error_code: str | None = None
):
    record = state_store.pop(state, None)
    if not record:
        raise HTTPException(status_code=400, detail='Invalid state')

    if status in ('success', 'reauthorized'):
        if not uid:
            raise HTTPException(status_code=400, detail='Missing uid')

        uid_map[record['internal_user_id']] = uid
        return RedirectResponse('/dashboard?connected=true')

    return RedirectResponse(f"/connect?error={error_code or 'unknown_error'}")


if __name__ == '__main__':
    uvicorn.run(app, host='0.0.0.0', port=3000)

Example 2: Understand data availability

You ask:

“What data can I get from Emerge and how soon after consent?”

AI response: The AI finds the Query Overview and Webhooks pages:

Data is available after export completes. Poll GET /export/status/{uid} and wait until the required provider in sources[] is data_ready: true.
Until a provider is ready, sync queries may return empty results. Handle this gracefully and retry with backoff.

Example 3: Debug a webhook signature issue

You ask:

“My webhook signature verification is failing. What am I doing wrong?”

AI response: The AI checks Webhooks and points out the most common issues:

Raw body required - sign the raw request body, not parsed JSON.
Header name - use the x-signature header.
Comparison - use constant-time comparison to avoid subtle mismatches.

import express from 'express';
import crypto from 'crypto';

const app = express();
app.use(express.raw({ type: 'application/json' }));

const WEBHOOK_SECRET = process.env.EMERGE_WEBHOOK_SECRET;
if (!WEBHOOK_SECRET) {
  throw new Error('Missing EMERGE_WEBHOOK_SECRET');
}

app.post('/webhooks/emerge', async (req, res) => {
  try {
    const signature = req.header('x-signature');
    if (!signature) {
      return res.status(401).send('Missing signature');
    }

    const expected = crypto
      .createHmac('sha256', WEBHOOK_SECRET)
      .update(req.body)
      .digest('hex');

    const signatureBuf = Buffer.from(signature, 'hex');
    const expectedBuf = Buffer.from(expected, 'hex');

    if (signatureBuf.length !== expectedBuf.length || !crypto.timingSafeEqual(signatureBuf, expectedBuf)) {
      return res.status(401).send('Invalid signature');
    }

    const payload = JSON.parse(req.body.toString('utf8'));
    const { event, uid } = payload;

    console.log(`Received ${event} for uid ${uid}`);
    return res.status(200).send('OK');
  } catch (err) {
    console.error('Webhook error', err);
    return res.status(500).send('Server error');
  }
});

app.listen(3001, () => {
  console.log('Webhook listener running on http://localhost:3001');
});

import os
import hmac
import hashlib
import json

from fastapi import FastAPI, HTTPException, Request
import uvicorn

app = FastAPI()

WEBHOOK_SECRET = os.getenv('EMERGE_WEBHOOK_SECRET')
if not WEBHOOK_SECRET:
    raise RuntimeError('Missing EMERGE_WEBHOOK_SECRET')


@app.post('/webhooks/emerge')
async def emerge_webhook(request: Request):
    signature = request.headers.get('x-signature')
    if not signature:
        raise HTTPException(status_code=401, detail='Missing signature')

    body = await request.body()
    expected = hmac.new(
        WEBHOOK_SECRET.encode(),
        body,
        hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(signature, expected):
        raise HTTPException(status_code=401, detail='Invalid signature')

    try:
        payload = json.loads(body)
    except json.JSONDecodeError as exc:
        raise HTTPException(status_code=400, detail='Invalid JSON') from exc

    event = payload.get('event')
    uid = payload.get('uid')
    print(f"Received {event} for uid {uid}")

    return {'status': 'ok'}


if __name__ == '__main__':
    uvicorn.run(app, host='0.0.0.0', port=3001)

Example 4: Plan a nightly sync

You ask:

“I want to sync Emerge data to my database nightly. What’s the best approach?”

AI response: The AI finds Pagination and Query Overview: Recommended approach:

Use delta queries with ingested_begin to fetch new data since the last run.
Page with cursor until has_more is false.
Store applied_ingested_end as your next ingested_begin.

interface SyncState {
  lastIngestedEnd: string;
}

async function deltaSync(uid: string, token: string, state: SyncState): Promise<SyncState> {
  let cursor: string | undefined;
  let appliedEnd = state.lastIngestedEnd;
  const newData: unknown[] = [];

  try {
    do {
      const params = new URLSearchParams({
        uid,
        ingested_begin: appliedEnd
      });

      if (cursor) {
        params.set('cursor', cursor);
      }

      const response = await fetch(
        `https://query.emergedata.ai/v1/sync/get_search?${params.toString()}`,
        { headers: { Authorization: `Bearer ${token}` } }
      );

      if (!response.ok) {
        const body = await response.text();
        throw new Error(`Query failed (${response.status}): ${body}`);
      }

      const result = await response.json();
      newData.push(...(result.data || []));

      appliedEnd = result.applied_ingested_end || appliedEnd;
      cursor = result.has_more ? result.next_cursor : undefined;
    } while (cursor);

    await processNewData(newData);
    return { lastIngestedEnd: appliedEnd };
  } catch (err) {
    console.error('Delta sync failed', err);
    throw err;
  }
}

async function processNewData(records: unknown[]) {
  console.log(`Processing ${records.length} records`);
}

async function runDeltaSync() {
  const token = process.env.EMERGE_API_TOKEN;
  if (!token) {
    throw new Error('Missing EMERGE_API_TOKEN');
  }

  const state = { lastIngestedEnd: '2024-01-01T00:00:00Z' };
  await deltaSync('psub_d4e5f6789012345678901234abcdef01', token, state);
}

runDeltaSync().catch(err => {
  console.error(err);
  process.exit(1);
});

import asyncio
import httpx
from dataclasses import dataclass
from typing import Any


@dataclass
class SyncState:
    last_ingested_end: str


async def delta_sync(uid: str, token: str, state: SyncState) -> SyncState:
    cursor: str | None = None
    applied_end = state.last_ingested_end
    new_data: list[Any] = []

    try:
        async with httpx.AsyncClient(timeout=30) as client:
            while True:
                params = {
                    'uid': uid,
                    'ingested_begin': applied_end
                }
                if cursor:
                    params['cursor'] = cursor

                response = await client.get(
                    'https://query.emergedata.ai/v1/sync/get_search',
                    params=params,
                    headers={'Authorization': f'Bearer {token}'}
                )
                response.raise_for_status()

                result = response.json()
                new_data.extend(result.get('data', []))

                applied_end = result.get('applied_ingested_end', applied_end)
                if not result.get('has_more'):
                    break
                cursor = result.get('next_cursor')

        await process_new_data(new_data)
        return SyncState(last_ingested_end=applied_end)
    except httpx.HTTPError as exc:
        raise RuntimeError(f'Delta sync failed: {exc}') from exc


async def process_new_data(records: list[Any]):
    print(f'Processing {len(records)} records')


if __name__ == '__main__':
    async def main():
        state = SyncState(last_ingested_end='2024-01-01T00:00:00Z')
        await delta_sync('psub_d4e5f6789012345678901234abcdef01', 'your_api_token', state)

    asyncio.run(main())

Example 5: Choose sync vs async for mobile

You ask:

“Should I use sync or async queries for my mobile app?”

AI response: For mobile apps, sync queries are recommended because they return JSON immediately and are simpler to integrate. Your mobile app should call your backend, not the Query API directly.

async function getSearchHistory(uid: string, token: string) {
  try {
    const response = await fetch(
      `https://query.emergedata.ai/v1/sync/get_search?uid=${encodeURIComponent(uid)}`,
      { headers: { Authorization: `Bearer ${token}` } }
    );

    if (!response.ok) {
      const body = await response.text();
      throw new Error(`Query failed (${response.status}): ${body}`);
    }

    return await response.json();
  } catch (err) {
    console.error('Search history query failed', err);
    throw err;
  }
}

async function runQuery() {
  const token = process.env.EMERGE_API_TOKEN;
  if (!token) {
    throw new Error('Missing EMERGE_API_TOKEN');
  }

  const data = await getSearchHistory('psub_d4e5f6789012345678901234abcdef01', token);
  console.log(data);
}

runQuery().catch(err => {
  console.error(err);
  process.exit(1);
});

import asyncio
import httpx


async def get_search_history(uid: str, token: str) -> dict:
    try:
        async with httpx.AsyncClient(timeout=30) as client:
            response = await client.get(
                'https://query.emergedata.ai/v1/sync/get_search',
                params={'uid': uid},
                headers={'Authorization': f'Bearer {token}'}
            )
            response.raise_for_status()
            return response.json()
    except httpx.HTTPError as exc:
        raise RuntimeError(f'Search history query failed: {exc}') from exc


if __name__ == '__main__':
    async def main():
        data = await get_search_history('psub_d4e5f6789012345678901234abcdef01', 'your_api_token')
        print(data)

    asyncio.run(main())

Example 6: Query user data via MCP

These examples show using the Query API MCP server to directly query consented user data from AI tools. This is different from the documentation MCP - it provides real data access.

Setup required: These examples require the Query API MCP server with valid credentials (auth_token and uid).

Analyze user interests from search history

Using Claude Code with Query API MCP configured:

claude "Analyze search patterns for psub_c3d4e5f6789012345678901234abcdef from the last month and identify the top 3 interest categories"

What happens:

Claude calls get_search_data tool with date range filter
Receives search history data grouped by category
Analyzes query frequency and patterns
Returns insights with specific examples

Example response:

Based on search history analysis:

1. **Technology/Software** (42% of searches)
   - Frequent queries about React, TypeScript, AWS
   - Pattern: Learning new frameworks

2. **Shopping/Electronics** (28% of searches)
   - Looking at laptop reviews, monitors, keyboards
   - Pattern: Office equipment research

3. **Health/Fitness** (18% of searches)
   - Running training plans, nutrition guides
   - Pattern: Marathon preparation

Generate personalized content recommendations

Using Cursor AI chat with Query API MCP configured:

"Based on this user's YouTube watch history and browsing patterns from the last 2 weeks,
suggest 5 relevant blog topics for our content calendar"

What happens:

AI calls get_youtube_data and get_browsing_data tools
Identifies themes and recurring interests
Cross-references patterns across data sources
Generates topic suggestions with rationale

Example response:

Recommended blog topics based on user data:

1. "Getting Started with Docker in 2024"
   - Watched 12 Docker tutorials, visited Docker docs 8 times

2. "Best Practices for React State Management"
   - Heavy focus on Redux and Zustand content

3. "Home Office Setup Guide for Developers"
   - Browsed ergonomic furniture, watched office tour videos

4. "Debugging Node.js Performance Issues"
   - Searched for profiling tools, read performance articles

5. "AI Tools for Code Review"
   - Interest in GitHub Copilot, Claude Code, and similar tools

Build user insights dashboard

Using Claude Desktop with Query API MCP:

"Create a user activity summary for the last 7 days showing:
- Most searched topics
- YouTube categories watched
- Top browsing domains
- Peak activity times"

What happens:

AI calls multiple MCP tools in parallel (get_search_data, get_youtube_data, get_browsing_data)
Aggregates and analyzes data across sources
Identifies patterns and trends
Formats results as structured summary

Example response:

User Activity Summary (Last 7 Days)

📊 Search Activity
- 127 total searches
- Top topics: Machine Learning (23), JavaScript (18), AWS (15)
- Peak search time: 10am-12pm EST

🎥 YouTube Activity
- 34 videos watched (8.2 hours)
- Top categories: Technology (41%), Education (32%), Entertainment (27%)
- Longest session: 2.5 hours on Sunday

🌐 Browsing Activity
- 412 page views
- Top domains: github.com (87), stackoverflow.com (64), docs.aws.amazon.com (42)
- Most active day: Wednesday (94 pages)

Delta sync for data warehouse

Using Claude Code to generate sync script:

claude "Write a Python script that uses the MCP tools to sync all new search data since my last sync timestamp into my data warehouse, handling pagination automatically"

What happens:

Claude accesses Query API MCP tool documentation
Generates script using get_search_data with delta query parameters
Implements pagination with cursor and has_more handling
Uses ingested_begin for incremental updates

Claude will generate a complete script that:

Loads last sync timestamp from state file
Calls MCP tool with ingested_begin parameter
Pages through all results
Saves applied_ingested_end for next sync

Try it yourself

For documentation help (Docs MCP):

“How do I implement retry logic for Emerge API calls?”
“What happens if a user revokes consent?”
“Generate a React component that shows a Connect Data button”
“Show the correct way to handle uid without asking the user”

See MCP Docs Setup to connect your AI tool.

For querying user data (Query API MCP):

“Show me search patterns for the last week”
“What categories is this user most interested in?”
“Compare browsing and YouTube activity to find common interests”
“Generate a weekly activity report”

See Query API MCP Setup to connect with credentials.

​Example 1: Generate integration code

​Example 2: Understand data availability

​Example 3: Debug a webhook signature issue

​Example 4: Plan a nightly sync

​Example 5: Choose sync vs async for mobile

​Example 6: Query user data via MCP

​Try it yourself

​For documentation help (Docs MCP):

​For querying user data (Query API MCP):

Example 1: Generate integration code

Example 2: Understand data availability

Example 3: Debug a webhook signature issue

Example 4: Plan a nightly sync

Example 5: Choose sync vs async for mobile

Example 6: Query user data via MCP

Try it yourself

For documentation help (Docs MCP):

For querying user data (Query API MCP):