Report Builder API Reference

Overview

The Report Builder API enables programmatic generation of AI-powered growth reports. Reports are generated asynchronously with optional real-time streaming of analysis progress.

Base URL: https://api.cogny.com/v1/reports

Authentication: Bearer token (see API Authentication)

Create Report

Generate a new AI-powered report from your data warehouse.

Endpoint

POST /v1/reports

Request Body

{
  "warehouse_id": "wh_123abc",
  "prompt": "Analyze conversion funnel performance for Q1 2025, identify drop-off points and recommend optimizations",
  "context": {
    "date_range": {
      "start": "2025-01-01",
      "end": "2025-03-31"
    },
    "dimensions": ["channel", "device_category", "country"],
    "focus_areas": ["conversion_rate", "revenue", "user_engagement"]
  },
  "options": {
    "include_visualizations": true,
    "include_sql_queries": true,
    "include_recommendations": true,
    "max_iterations": 50,
    "language": "en"
  },
  "metadata": {
    "created_by": "user_456",
    "department": "growth",
    "project": "q1-optimization"
  }
}

Parameters

Response

{
  "success": true,
  "data": {
    "id": "rpt_xyz789",
    "warehouse_id": "wh_123abc",
    "status": "processing",
    "prompt": "Analyze conversion funnel performance for Q1 2025...",
    "created_at": "2025-02-05T10:30:00Z",
    "updated_at": "2025-02-05T10:30:00Z",
    "estimated_completion": "2025-02-05T10:33:00Z",
    "url": "https://app.cogny.com/reports/rpt_xyz789",
    "stream_url": "https://api.cogny.com/v1/reports/rpt_xyz789/stream"
  }
}

Example Request (cURL)

curl -X POST https://api.cogny.com/v1/reports \
  -H "Authorization: Bearer sk_live_abc123xyz789" \
  -H "Content-Type: application/json" \
  -d '{
    "warehouse_id": "wh_123abc",
    "prompt": "Analyze conversion funnel performance for Q1 2025",
    "options": {
      "include_visualizations": true,
      "include_recommendations": true
    }
  }'

Example Request (Python)

import requests

API_KEY = "sk_live_abc123xyz789"
BASE_URL = "https://api.cogny.com/v1"

headers = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json"
}

payload = {
    "warehouse_id": "wh_123abc",
    "prompt": "Analyze conversion funnel performance for Q1 2025",
    "context": {
        "date_range": {
            "start": "2025-01-01",
            "end": "2025-03-31"
        },
        "focus_areas": ["conversion_rate", "revenue"]
    },
    "options": {
        "include_visualizations": True,
        "include_recommendations": True
    }
}

response = requests.post(
    f"{BASE_URL}/reports",
    headers=headers,
    json=payload
)

report = response.json()["data"]
print(f"Report created: {report['id']}")
print(f"Status: {report['status']}")
print(f"Stream URL: {report['stream_url']}")

Get Report Status

Retrieve current status and details of a report.

Endpoint

GET /v1/reports/{report_id}

Response

{
  "success": true,
  "data": {
    "id": "rpt_xyz789",
    "warehouse_id": "wh_123abc",
    "status": "completed",
    "prompt": "Analyze conversion funnel performance for Q1 2025...",
    "created_at": "2025-02-05T10:30:00Z",
    "updated_at": "2025-02-05T10:33:45Z",
    "completed_at": "2025-02-05T10:33:45Z",
    "url": "https://app.cogny.com/reports/rpt_xyz789",
    "summary": {
      "total_queries": 12,
      "total_insights": 8,
      "key_findings": [
        "Mobile conversion rate 23% lower than desktop",
        "Cart abandonment peaks at payment step (41%)",
        "Organic traffic has highest conversion rate (4.2%)"
      ]
    },
    "metadata": {
      "created_by": "user_456",
      "department": "growth"
    }
  }
}

Status Values

Example (Python with Polling)

import time

def wait_for_report(report_id, timeout=300, interval=5):
    """Poll report status until completion"""
    start_time = time.time()

    while time.time() - start_time < timeout:
        response = requests.get(
            f"{BASE_URL}/reports/{report_id}",
            headers=headers
        )

        report = response.json()["data"]
        status = report["status"]

        print(f"Status: {status}")

        if status == "completed":
            return report
        elif status == "failed":
            raise Exception(f"Report failed: {report.get('error')}")

        time.sleep(interval)

    raise TimeoutError("Report generation timed out")

# Usage
report = wait_for_report("rpt_xyz789")
print(f"Report completed: {report['url']}")

Stream Report Generation

Stream real-time updates during report generation using Server-Sent Events (SSE).

Endpoint

GET /v1/reports/{report_id}/stream

SSE Event Types

status Event

Status updates during processing:

event: status
data: {"status": "processing", "message": "Analyzing conversion funnel data..."}

query Event

SQL query execution updates:

event: query
data: {"query": "SELECT event_name, COUNT(*) as count...", "status": "executing"}

result Event

Query results:

event: result
data: {"rows": 1247, "columns": ["event_name", "count", "conversion_rate"]}

insight Event

AI-generated insights:

event: insight
data: {"type": "finding", "content": "Mobile conversion rate is 23% lower than desktop", "confidence": 0.92}

visualization Event

Chart or graph generated:

event: visualization
data: {"type": "line_chart", "title": "Conversion Rate by Channel", "url": "https://cdn.cogny.com/viz_123.png"}

completed Event

Report generation finished:

event: completed
data: {"report_id": "rpt_xyz789", "url": "https://app.cogny.com/reports/rpt_xyz789"}

error Event

Error occurred:

event: error
data: {"code": "QUERY_FAILED", "message": "Insufficient permissions to access table"}

Example (Python with SSE)

import requests
import json

def stream_report(report_id):
    """Stream report generation progress"""
    url = f"{BASE_URL}/reports/{report_id}/stream"

    with requests.get(url, headers=headers, stream=True) as response:
        response.raise_for_status()

        for line in response.iter_lines():
            if not line:
                continue

            line = line.decode('utf-8')

            # Parse SSE format
            if line.startswith('event:'):
                event_type = line.split(':', 1)[1].strip()
            elif line.startswith('data:'):
                data = json.loads(line.split(':', 1)[1].strip())

                if event_type == 'status':
                    print(f"[STATUS] {data['message']}")
                elif event_type == 'insight':
                    print(f"[INSIGHT] {data['content']}")
                elif event_type == 'completed':
                    print(f"[COMPLETED] {data['url']}")
                    return data
                elif event_type == 'error':
                    raise Exception(f"Error: {data['message']}")

# Usage
report = stream_report("rpt_xyz789")

Example (JavaScript/TypeScript)

async function streamReport(reportId: string) {
  const response = await fetch(
    `${BASE_URL}/reports/${reportId}/stream`,
    {
      headers: {
        'Authorization': `Bearer ${API_KEY}`,
      },
    }
  );

  const reader = response.body?.getReader();
  const decoder = new TextDecoder();

  while (true) {
    const { done, value } = await reader!.read();
    if (done) break;

    const chunk = decoder.decode(value);
    const lines = chunk.split('\n');

    let eventType = '';

    for (const line of lines) {
      if (line.startsWith('event:')) {
        eventType = line.substring(6).trim();
      } else if (line.startsWith('data:')) {
        const data = JSON.parse(line.substring(5).trim());

        switch (eventType) {
          case 'status':
            console.log(`[STATUS] ${data.message}`);
            break;
          case 'insight':
            console.log(`[INSIGHT] ${data.content}`);
            break;
          case 'completed':
            console.log(`[COMPLETED] ${data.url}`);
            return data;
          case 'error':
            throw new Error(`Error: ${data.message}`);
        }
      }
    }
  }
}

// Usage
const report = await streamReport('rpt_xyz789');

List Reports

Retrieve all reports for a warehouse with filtering and pagination.

Endpoint

GET /v1/reports

Query Parameters

Response

{
  "success": true,
  "data": [
    {
      "id": "rpt_xyz789",
      "warehouse_id": "wh_123abc",
      "status": "completed",
      "prompt": "Analyze conversion funnel performance...",
      "created_at": "2025-02-05T10:30:00Z",
      "completed_at": "2025-02-05T10:33:45Z",
      "url": "https://app.cogny.com/reports/rpt_xyz789"
    }
  ],
  "pagination": {
    "has_more": true,
    "next_cursor": "eyJpZCI6InJwdF94eXo3ODkifQ",
    "total_count": 47
  }
}

Example

curl -X GET "https://api.cogny.com/v1/reports?warehouse_id=wh_123abc&status=completed&limit=50" \
  -H "Authorization: Bearer sk_live_abc123xyz789"

Get Report Content

Retrieve the full content of a completed report.

Endpoint

GET /v1/reports/{report_id}/content

Response

{
  "success": true,
  "data": {
    "report_id": "rpt_xyz789",
    "sections": [
      {
        "title": "Executive Summary",
        "content": "Analysis of Q1 2025 conversion funnel reveals...",
        "insights": [
          {
            "type": "finding",
            "content": "Mobile conversion rate is 23% lower than desktop",
            "confidence": 0.92,
            "data": {
              "mobile_rate": 0.027,
              "desktop_rate": 0.035
            }
          }
        ]
      },
      {
        "title": "Funnel Analysis",
        "content": "The conversion funnel shows significant drop-off...",
        "visualizations": [
          {
            "type": "funnel_chart",
            "title": "Conversion Funnel by Step",
            "url": "https://cdn.cogny.com/viz_123.png",
            "data": {
              "steps": [
                {"name": "Landing", "users": 100000, "rate": 1.0},
                {"name": "Product View", "users": 45000, "rate": 0.45},
                {"name": "Add to Cart", "users": 12000, "rate": 0.12},
                {"name": "Checkout", "users": 7200, "rate": 0.072},
                {"name": "Purchase", "users": 3500, "rate": 0.035}
              ]
            }
          }
        ],
        "queries": [
          {
            "sql": "SELECT event_name, COUNT(DISTINCT user_pseudo_id) as users...",
            "description": "Calculate funnel step conversion rates",
            "rows_returned": 5,
            "execution_time_ms": 1247
          }
        ]
      }
    ],
    "recommendations": [
      {
        "priority": "high",
        "title": "Optimize Mobile Checkout Flow",
        "description": "Mobile users abandon cart at 2x desktop rate...",
        "estimated_impact": "12-18% increase in mobile conversions",
        "effort": "medium",
        "action_items": [
          "Implement one-click checkout for returning users",
          "Reduce form fields from 12 to 6",
          "Add mobile payment options (Apple Pay, Google Pay)"
        ]
      }
    ],
    "metadata": {
      "total_queries": 12,
      "total_insights": 8,
      "generation_time_seconds": 225
    }
  }
}

Cancel Report

Cancel a queued or processing report.

Endpoint

DELETE /v1/reports/{report_id}

Response

{
  "success": true,
  "data": {
    "id": "rpt_xyz789",
    "status": "cancelled",
    "cancelled_at": "2025-02-05T10:32:00Z"
  }
}

Error Codes

Best Practices

1. Provide Context

Include relevant context to improve report quality:

payload = {
    "warehouse_id": "wh_123abc",
    "prompt": "Analyze conversion funnel performance",
    "context": {
        "date_range": {
            "start": "2025-01-01",
            "end": "2025-03-31"
        },
        "dimensions": ["channel", "device_category"],
        "focus_areas": ["conversion_rate", "revenue"],
        "business_context": "Launched new checkout flow on Feb 1st"
    }
}

2. Use Streaming for Real-Time Feedback

Stream progress for long-running reports:

# Better user experience with streaming
report_data = stream_report(report_id)

# vs. polling
report_data = wait_for_report(report_id)

3. Handle Errors Gracefully

Implement proper error handling:

try:
    report = create_report(warehouse_id, prompt)
    result = wait_for_report(report['id'])
except requests.exceptions.HTTPError as e:
    if e.response.status_code == 429:
        print("Rate limit exceeded. Please wait.")
    elif e.response.status_code == 400:
        error = e.response.json()
        print(f"Invalid request: {error['error']['message']}")
    else:
        print(f"API error: {e}")

4. Store Report IDs

Save report IDs for future reference:

# Store in your database
db.save_report({
    'cogny_report_id': report['id'],
    'user_id': user_id,
    'prompt': prompt,
    'status': report['status'],
    'url': report['url']
})

Next Steps

• Growth Tickets API - Access AI-generated growth recommendations
• BigQuery Setup - Connect your data warehouse
• API Authentication - Learn about authentication and rate limits