> ## Documentation Index
> Fetch the complete documentation index at: https://docs.emergedata.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# AI Workflow Examples

> Real-world examples of using AI tools with Emerge documentation

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](/link/create-link) and [Callbacks](/link/callbacks) pages, and generates code that keeps `uid` server-side (never user input):

<CodeGroup>
  ```typescript TypeScript theme={null}
  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');
  });
  ```

  ```python Python theme={null}
  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)
  ```
</CodeGroup>

## 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](/query/overview) and [Webhooks](/link/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](/link/webhooks) and points out the most common issues:

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

<CodeGroup>
  ```typescript TypeScript theme={null}
  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');
  });
  ```

  ```python Python theme={null}
  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)
  ```
</CodeGroup>

## 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](/query/pagination) and [Query Overview](/query/overview):

```mermaid theme={null}
graph LR
    A[Cron Job] --> B[Delta Query]
    B --> C[Process Records]
    C --> D[Update Database]
    D --> E[Store Sync State]
```

**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`.

<CodeGroup>
  ```typescript TypeScript theme={null}
  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);
  });
  ```

  ```python Python theme={null}
  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())
  ```
</CodeGroup>

## 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.

<CodeGroup>
  ```typescript TypeScript theme={null}
  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);
  });
  ```

  ```python Python theme={null}
  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())
  ```
</CodeGroup>

## 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.

<Note>
  **Setup required:** These examples require the [Query API MCP server](/ai/mcp-query-setup) with valid credentials (`auth_token` and `uid`).
</Note>

<AccordionGroup>
  <Accordion title="Analyze user interests from search history">
    Using Claude Code with Query API MCP configured:

    ```bash theme={null}
    claude "Analyze search patterns for psub_c3d4e5f6789012345678901234abcdef from the last month and identify the top 3 interest categories"
    ```

    **What happens:**

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

    **Example response:**

    ```text theme={null}
    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
    ```
  </Accordion>

  <Accordion title="Generate personalized content recommendations">
    Using Cursor AI chat with Query API MCP configured:

    ```text theme={null}
    "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:**

    1. AI calls `get_youtube_data` and `get_browsing_data` tools
    2. Identifies themes and recurring interests
    3. Cross-references patterns across data sources
    4. Generates topic suggestions with rationale

    **Example response:**

    ```text theme={null}
    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
    ```
  </Accordion>

  <Accordion title="Build user insights dashboard">
    Using Claude Desktop with Query API MCP:

    ```text theme={null}
    "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:**

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

    **Example response:**

    ```text theme={null}
    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)
    ```
  </Accordion>

  <Accordion title="Delta sync for data warehouse">
    Using Claude Code to generate sync script:

    ```bash theme={null}
    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:**

    1. Claude accesses Query API MCP tool documentation
    2. Generates script using `get_search_data` with delta query parameters
    3. Implements pagination with `cursor` and `has_more` handling
    4. 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
  </Accordion>
</AccordionGroup>

## 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](/ai/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](/ai/mcp-query-setup) to connect with credentials.