How to Integrate Replit with Typeform

1import os
2import requests
3from typing import Optional
4
5TOKEN = os.environ["TYPEFORM_TOKEN"]
6BASE_URL = "https://api.typeform.com"
7
8HEADERS = {
9    "Authorization": f"Bearer {TOKEN}",
10    "Content-Type": "application/json"
11}
12
13def list_forms() -> list:
14    """Get all forms in the Typeform account."""
15    response = requests.get(f"{BASE_URL}/forms", headers=HEADERS)
16    response.raise_for_status()
17    return response.json().get("items", [])
18
19def get_form(form_id: str) -> dict:
20    """Get the full definition of a form including field names."""
21    response = requests.get(f"{BASE_URL}/forms/{form_id}", headers=HEADERS)
22    response.raise_for_status()
23    return response.json()
24
25def get_responses(form_id: str, page_size: int = 25,
26                  since: str = None, until: str = None) -> dict:
27    """Fetch form responses with optional date range filtering."""
28    params = {"page_size": page_size}
29    if since:
30        params["since"] = since  # ISO 8601 datetime string
31    if until:
32        params["until"] = until
33
34    response = requests.get(
35        f"{BASE_URL}/forms/{form_id}/responses",
36        params=params,
37        headers=HEADERS
38    )
39    response.raise_for_status()
40    return response.json()
41
42def build_field_map(form: dict) -> dict:
43    """Build a map from field ID to field title for answer parsing."""
44    field_map = {}
45    for field in form.get("fields", []):
46        field_map[field["id"]] = field.get("title", field["id"])
47        # Handle group fields (pages/sets)
48        for sub_field in field.get("properties", {}).get("fields", []):
49            field_map[sub_field["id"]] = sub_field.get("title", sub_field["id"])
50    return field_map
51
52def parse_answer(answer: dict) -> any:
53    """Extract the value from a Typeform answer regardless of type."""
54    answer_type = answer.get("type")
55    if answer_type == "text":
56        return answer.get("text")
57    elif answer_type == "email":
58        return answer.get("email")
59    elif answer_type == "number":
60        return answer.get("number")
61    elif answer_type == "choice":
62        return answer.get("choice", {}).get("label")
63    elif answer_type == "choices":
64        return [c.get("label") for c in answer.get("choices", {}).get("labels", [])]
65    elif answer_type == "boolean":
66        return answer.get("boolean")
67    elif answer_type == "date":
68        return answer.get("date")
69    elif answer_type == "file_url":
70        return answer.get("file_url")
71    return answer.get(answer_type)  # Fallback for other types
72
73def responses_to_dicts(form_id: str) -> list:
74    """Convert raw API responses into flat dicts with field names as keys."""
75    form = get_form(form_id)
76    field_map = build_field_map(form)
77    responses = get_responses(form_id)
78    
79    results = []
80    for item in responses.get("items", []):
81        response_dict = {
82            "response_id": item["response_id"],
83            "submitted_at": item["submitted_at"],
84            "email": item.get("hidden", {}).get("email", "")  # Hidden fields
85        }
86        for answer in item.get("answers", []):
87            field_id = answer.get("field", {}).get("id", "")
88            field_title = field_map.get(field_id, field_id)
89            response_dict[field_title] = parse_answer(answer)
90        results.append(response_dict)
91    return results
92
93# Example usage
94if __name__ == "__main__":
95    forms = list_forms()
96    print(f"Forms in account: {len(forms)}")
97    for form in forms:
98        print(f"  {form['id']}: {form['title']}")
99
100    if forms:
101        form_id = forms[0]["id"]
102        responses = responses_to_dicts(form_id)
103        print(f"\nLatest {len(responses)} responses for '{forms[0]['title']}':")
104        for r in responses[:3]:
105            print(f"  {r['submitted_at']}: {r}")

1const express = require('express');
2const axios = require('axios');
3const crypto = require('crypto');
4
5const app = express();
6
7const TOKEN = process.env.TYPEFORM_TOKEN;
8const WEBHOOK_SECRET = process.env.TYPEFORM_SECRET;
9const BASE_URL = 'https://api.typeform.com';
10
11const typeform = axios.create({
12  baseURL: BASE_URL,
13  headers: { 'Authorization': `Bearer ${TOKEN}` }
14});
15
16function verifySignature(rawBody, signature) {
17  if (!WEBHOOK_SECRET) return true; // Skip if not configured
18  const expected = 'sha256=' + crypto
19    .createHmac('sha256', WEBHOOK_SECRET)
20    .update(rawBody)
21    .digest('base64'); // Note: Typeform uses base64, not hex
22  return crypto.timingSafeEqual(
23    Buffer.from(expected),
24    Buffer.from(signature || '')
25  );
26}
27
28function parseAnswer(answer) {
29  const type = answer.type;
30  switch (type) {
31    case 'text': return answer.text;
32    case 'email': return answer.email;
33    case 'number': return answer.number;
34    case 'choice': return answer.choice?.label;
35    case 'choices': return answer.choices?.labels || [];
36    case 'boolean': return answer.boolean;
37    case 'date': return answer.date;
38    default: return answer[type];
39  }
40}
41
42// Webhook endpoint — receives form submissions in real time
43app.post('/typeform/webhook',
44  express.raw({ type: 'application/json' }),
45  (req, res) => {
46    const signature = req.headers['typeform-signature'];
47    if (!verifySignature(req.body, signature)) {
48      console.warn('Invalid webhook signature — rejecting');
49      return res.status(403).json({ error: 'Invalid signature' });
50    }
51
52    const data = JSON.parse(req.body.toString());
53    const formId = data.form_response?.form_id;
54    const responseId = data.form_response?.token;
55    const submittedAt = data.form_response?.submitted_at;
56    const answers = data.form_response?.answers || [];
57    const hidden = data.form_response?.hidden || {};
58
59    console.log(`New submission on form ${formId} (${responseId}) at ${submittedAt}`);
60    console.log('Hidden fields:', hidden);
61
62    // Parse answers into a usable object
63    const parsed = {};
64    for (const answer of answers) {
65      const fieldId = answer.field?.id || 'unknown';
66      parsed[fieldId] = parseAnswer(answer);
67    }
68    console.log('Parsed answers:', parsed);
69
70    // TODO: send to CRM, database, Slack, etc.
71    // Example: send to your own /crm/contact endpoint
72    // axios.post('/crm/contact', { email: parsed.email, ...parsed });
73
74    res.json({ received: true });
75  }
76);
77
78// Get forms list
79app.get('/forms', async (req, res) => {
80  try {
81    const { data } = await typeform.get('/forms');
82    res.json(data.items || []);
83  } catch (err) {
84    res.status(500).json({ error: err.message });
85  }
86});
87
88// Get latest responses for a form
89app.get('/forms/:formId/responses', async (req, res) => {
90  try {
91    const { data } = await typeform.get(`/forms/${req.params.formId}/responses`, {
92      params: { page_size: 25 }
93    });
94    res.json(data);
95  } catch (err) {
96    res.status(500).json({ error: err.message });
97  }
98});
99
100// Health check (must use regular JSON body parsing for non-webhook routes)
101app.use(express.json());
102app.get('/health', (req, res) => res.json({ status: 'ok' }));
103
104app.listen(3000, '0.0.0.0', () => {
105  console.log('Typeform integration server running on port 3000');
106});

1import os
2import requests
3
4TOKEN = os.environ["TYPEFORM_TOKEN"]
5WEBHOOK_SECRET = os.environ["TYPEFORM_SECRET"]
6BASE_URL = "https://api.typeform.com"
7
8HEADERS = {
9    "Authorization": f"Bearer {TOKEN}",
10    "Content-Type": "application/json"
11}
12
13def register_webhook(form_id: str, webhook_url: str,
14                     tag: str = "replit-webhook") -> dict:
15    """Register a webhook for a specific Typeform form."""
16    payload = {
17        "tag": tag,
18        "url": webhook_url,
19        "enabled": True,
20        "secret": WEBHOOK_SECRET,
21        "verify_ssl": True
22    }
23    response = requests.post(
24        f"{BASE_URL}/forms/{form_id}/webhooks/{tag}",
25        json=payload,
26        headers=HEADERS
27    )
28    response.raise_for_status()
29    return response.json()
30
31def list_webhooks(form_id: str) -> list:
32    """List all webhooks registered for a form."""
33    response = requests.get(
34        f"{BASE_URL}/forms/{form_id}/webhooks",
35        headers=HEADERS
36    )
37    response.raise_for_status()
38    return response.json().get("items", [])
39
40def delete_webhook(form_id: str, tag: str) -> None:
41    """Remove a webhook from a form."""
42    response = requests.delete(
43        f"{BASE_URL}/forms/{form_id}/webhooks/{tag}",
44        headers=HEADERS
45    )
46    response.raise_for_status()
47
48# Example: register webhook for a form
49if __name__ == "__main__":
50    form_id = "YOUR_FORM_ID"  # Replace with your form ID
51    webhook_url = "https://your-app.replit.app/typeform/webhook"
52
53    result = register_webhook(form_id, webhook_url)
54    print(f"Webhook registered: {result}")
55
56    webhooks = list_webhooks(form_id)
57    print(f"\nAll webhooks for form {form_id}:")
58    for wh in webhooks:
59        print(f"  {wh['tag']}: {wh['url']} (enabled: {wh['enabled']})")