Telegram Bot CRM Integration: Connect to HubSpot, Salesforce & More

 1pip install aiogram httpx

 1# Telegram bot integration with HubSpot CRM
 2# Creates a contact in HubSpot when form is filled in bot
 3
 4import httpx
 5from aiogram import Router, F
 6from aiogram.types import Message
 7from aiogram.fsm.context import FSMContext
 8from aiogram.fsm.state import State, StatesGroup
 9
10router = Router()
11
12# ─── HubSpot settings ───
13HUBSPOT_API_KEY = "your-hubspot-private-app-token"
14HUBSPOT_BASE_URL = "https://api.hubapi.com"
15
16# ─── Form states (FSM) ───
17class LeadForm(StatesGroup):
18    """States for collecting lead data through bot."""
19    waiting_name = State()
20    waiting_email = State()
21    waiting_interest = State()
22
23@router.message(LeadForm.waiting_name)
24async def process_name(message: Message, state: FSMContext):
25    """Save name and ask for email."""
26    await state.update_data(name=message.text)
27    await message.answer("What is your email address?")
28    await state.set_state(LeadForm.waiting_email)
29
30@router.message(LeadForm.waiting_email)
31async def process_email(message: Message, state: FSMContext):
32    """Save email and ask for interest."""
33    await state.update_data(email=message.text)
34    await message.answer(
35        "What are you interested in?\n"
36        "1️⃣ Bot development\n"
37        "2️⃣ CRM integration\n"
38        "3️⃣ E-commerce bot"
39    )
40    await state.set_state(LeadForm.waiting_interest)
41
42@router.message(LeadForm.waiting_interest)
43async def process_interest(message: Message, state: FSMContext):
44    """
45    Collect all data and send to HubSpot.
46    This is the final form step.
47    """
48    data = await state.get_data()
49    interest_map = {
50        "1": "Bot development",
51        "2": "CRM integration",
52        "3": "E-commerce bot",
53    }
54    interest = interest_map.get(message.text, message.text)
55
56    # ─── Create contact in HubSpot ───
57    hubspot_result = await create_hubspot_contact(
58        name=data["name"],
59        email=data["email"],
60        interest=interest,
61        telegram_username=message.from_user.username,
62    )
63
64    if hubspot_result:
65        await message.answer(
66            f"✅ Thanks, {data['name']}! "
67            f"Our team will reach out within 24 hours."
68        )
69    else:
70        await message.answer(
71            f"✅ Thanks, {data['name']}! "
72            f"We have received your request."
73        )
74
75    await state.clear()
76
77async def create_hubspot_contact(
78    name: str,
79    email: str,
80    interest: str,
81    telegram_username: str | None = None,
82) -> bool:
83    """
84    Creates or updates contact in HubSpot.
85    Returns True on success, False on error.
86    """
87    # Split name into first/last (HubSpot requires separate fields)
88    name_parts = name.strip().split(" ", 1)
89    first_name = name_parts[0]
90    last_name = name_parts[1] if len(name_parts) > 1 else ""
91
92    # Contact data for HubSpot API
93    properties = {
94        "email": email,
95        "firstname": first_name,
96        "lastname": last_name,
97        "lead_source": "Telegram Bot",
98        "hs_lead_status": "NEW",
99        # Custom field — must be created in HubSpot beforehand
100        "telegram_username": telegram_username or "",
101        "interested_in": interest,
102    }
103
104    headers = {
105        "Authorization": f"Bearer {HUBSPOT_API_KEY}",
106        "Content-Type": "application/json",
107    }
108
109    async with httpx.AsyncClient() as client:
110        try:
111            # HubSpot automatically deduplicates by email
112            response = await client.post(
113                f"{HUBSPOT_BASE_URL}/crm/v3/objects/contacts",
114                headers=headers,
115                json={"properties": properties},
116                timeout=10.0,
117            )
118
119            if response.status_code in (200, 201):
120                contact_id = response.json().get("id")
121                print(f"HubSpot contact created: {contact_id}")
122                return True
123            elif response.status_code == 409):
124                # Contact already exists — updating
125                print("Contact exists, updating...")
126                return await update_hubspot_contact(email, properties)
127            else:
128                print(f"HubSpot error: {response.status_code}")
129                print(response.text)
130                return False
131
132        except Exception as e:
133            print(f"HubSpot API error: {e}")
134            return False
135
136async def update_hubspot_contact(email: str, properties: dict) -> bool:
137    """Updates existing contact in HubSpot by email."""
138    headers = {
139        "Authorization": f"Bearer {HUBSPOT_API_KEY}",
140        "Content-Type": "application/json",
141    }
142    async with httpx.AsyncClient() as client:
143        try:
144            response = await client.patch(
145                f"{HUBSPOT_BASE_URL}/crm/v3/objects/contacts/{email}",
146                headers=headers,
147                json={"properties": properties},
148                params={"idProperty": "email"},
149                timeout=10.0,
150            )
151            return response.status_code == 200
152        except Exception as e:
153            print(f"HubSpot update error: {e}")
154            return False

 1# Salesforce OAuth 2.0 authentication
 2# Using "Client Credentials" flow for server applications
 3
 4import httpx
 5
 6SALESFORCE_INSTANCE = "https://your-instance.salesforce.com"
 7CLIENT_ID = "your-connected-app-client-id"
 8CLIENT_SECRET = "your-connected-app-client-secret"
 9
10async def get_salesforce_token() -> str | None:
11    """
12    Gets access token for Salesforce API.
13    Token lives ~2 hours, needs caching and refreshing.
14    """
15    async with httpx.AsyncClient() as client:
16        response = await client.post(
17            f"{SALESFORCE_INSTANCE}/services/oauth2/token",
18            data={
19                "grant_type": "client_credentials",
20                "client_id": CLIENT_ID,
21                "client_secret": CLIENT_SECRET,
22            },
23        )
24        if response.status_code == 200:
25            return response.json()["access_token"]
26        return None
27
28async def create_salesforce_lead(
29    token: str,
30    first_name: str,
31    last_name: str,
32    email: str,
33    phone: str,
34    source: str = "Telegram Bot",
35    interest: str = "",
36) -> str | None:
37    """
38    Creates Lead in Salesforce.
39    Returns ID of created lead.
40    """
41    headers = {
42        "Authorization": f"Bearer {token}",
43        "Content-Type": "application/json",
44    }
45
46    lead_data = {
47        "FirstName": first_name,
48        "LastName": last_name,
49        "Email": email,
50        "Phone": phone,
51        "LeadSource": source,
52        "Description": f"Interest: {interest}. Source: Telegram bot.",
53        # Custom field (create in Salesforce Setup → Object Manager)
54        "Telegram_Username__c": "",
55    }
56
57    async with httpx.AsyncClient() as client:
58        response = await client.post(
59            f"{SALESFORCE_INSTANCE}/services/data/v59.0/sobjects/Lead",
60            headers=headers,
61            json=lead_data,
62            timeout=10.0,
63        )
64
65        if response.status_code == 201:
66            lead_id = response.json()["id"]
67            print(f"Salesforce Lead created: {lead_id}")
68            return lead_id
69        else:
70            print(f"Salesforce error: {response.text}")
71            return None

 1# Telegram bot integration with Pipedrive CRM
 2# Creates Person and Deal when form is filled
 3
 4import httpx
 5
 6PIPEDRIVE_API_TOKEN = "your-pipedrive-api-token"
 7PIPEDRIVE_BASE_URL = "https://api.pipedrive.com/v1"
 8
 9async def create_pipedrive_person(
10    name: str,
11    email: str,
12    phone: str = "",
13    telegram_username: str = "",
14) -> int | None:
15    """
16    Creates a Person in Pipedrive. Returns ID of created contact.
17    """
18    headers = {"Content-Type": "application/json"}
19    params = {"api_token": PIPEDRIVE_API_TOKEN}
20
21    person_data = {
22        "name": name,
23        "email": [{"value": email, "primary": True}],
24        "phone": [{"value": phone, "primary": True}] if phone else [],
25        # Custom field for Telegram (create in Settings → Data fields)
26        "4b7fa47a7e1c1c2c": telegram_username,  # Example custom field ID
27    }
28
29    async with httpx.AsyncClient() as client:
30        response = await client.post(
31            f"{PIPEDRIVE_BASE_URL}/persons",
32            headers=headers,
33            params=params,
34            json=person_data,
35            timeout=10.0,
36        )
37
38        if response.status_code == 201:
39            person_id = response.json()["data"]["id"]
40            print(f"Pipedrive Person created: {person_id}")
41            return person_id
42        else:
43            print(f"Pipedrive error: {response.text}")
44            return None
45
46async def create_pipedrive_deal(
47    person_id: int,
48    title: str,
49    value: float = 0,
50    pipeline_id: int = 0,
51    stage_id: int = 1,
52) -> int | None:
53    """
54    Creates a Deal in Pipedrive, linked to contact.
55    """
56    headers = {"Content-Type": "application/json"}
57    params = {"api_token": PIPEDRIVE_API_TOKEN}
58
59    deal_data = {
60        "title": title,
61        "person_id": person_id,
62        "value": value,
63        "currency": "USD",
64    }
65
66    # If pipeline and stage are specified
67    if pipeline_id:
68        deal_data["pipeline_id"] = pipeline_id
69    if stage_id:
70        deal_data["stage_id"] = stage_id
71
72    async with httpx.AsyncClient() as client:
73        response = await client.post(
74            f"{PIPEDRIVE_BASE_URL}/deals",
75            headers=headers,
76            params=params,
77            json=deal_data,
78            timeout=10.0,
79        )
80
81        if response.status_code == 201:
82            deal_id = response.json()["data"]["id"]
83            print(f"Pipedrive Deal created: {deal_id}")
84            return deal_id
85        else:
86            print(f"Pipedrive error: {response.text}")
87            return None
88
89async def add_pipedrive_activity(
90    deal_id: int,
91    subject: str,
92    note: str,
93    due_date: str,
94) -> bool:
95    """
96    Adds activity (follow-up task) to deal. Used to remind manager to call back.
97    """
98    headers = {"Content-Type": "application/json"}
99    params = {"api_token": PIPEDRIVE_API_TOKEN}
100
101    activity_data = {
102        "subject": subject,
103        "type": "call",
104        "due_date": due_date,
105        "deal_id": deal_id,
106        "note": note,
107    }
108
109    async with httpx.AsyncClient() as client:
110        response = await client.post(
111            f"{PIPEDRIVE_BASE_URL}/activities",
112            headers=headers,
113            params=params,
114            json=activity_data,
115            timeout=10.0,
116        )
117        return response.status_code == 201

 1# Sending lead data to Zapier webhook
 2# Zapier will redirect data to your CRM
 3
 4import httpx
 5from datetime import datetime
 6
 7# Webhook URL from Zapier (created in Zapier → Create → Zap → Webhooks by Zapier)
 8ZAPIER_WEBHOOK_URL = "https://hooks.zapier.com/hooks/catch/123456/abcdef/"
 9
10async def send_to_zapier(
11    name: str,
12    email: str,
13    phone: str = "",
14    interest: str = "",
15    telegram_username: str = "",
16    chat_id: int = 0,
17) -> bool:
18    """
19    Sends lead data to Zapier via webhook. Zapier will automatically redirect to CRM.
20    """
21    payload = {
22        "name": name,
23        "email": email,
24        "phone": phone,
25        "interest": interest,
26        "telegram_username": telegram_username,
27        "chat_id": str(chat_id),
28        "source": "telegram_bot",
29        "timestamp": datetime.utcnow().isoformat(),
30    }
31
32    async with httpx.AsyncClient() as client:
33        try:
34            response = await client.post(
35                ZAPIER_WEBHOOK_URL,
36                json=payload,
37                timeout=10.0,
38            )
39            success = response.status_code == 200
40            if success:
41                print(f"Zapier webhook sent for {name}")
42            else:
43                print(f"Zapier error: {response.status_code}")
44            return success
45        except Exception as e:
46            print(f"Zapier webhook error: {e}")
47            return False

 1async def generate_conversation_summary(messages: list[dict]) -> str:
 2    """
 3    Generates a brief conversation summary for CRM. In a real project, LLM can be used for summarization. For now — simple keyword-based version.
 4    """
 5    # Collect all user messages
 6    user_messages = [
 7        m["text"] for m in messages if m["role"] == "user"
 8    ]
 9
10    if not user_messages:
11        return "User started conversation but did not provide details."
12
13    # Determine topic by keywords
14    interests = []
15    text_blob = " ".join(user_messages).lower()
16
17    keywords_map = {
18        "pricing": "pricing inquiry",
19        "price": "pricing inquiry",
20        "how much": "pricing inquiry",
21        "demo": "demo request",
22        "trial": "trial request",
23        "integrate": "integration question",
24        "api": "technical question",
25        "custom": "custom solution inquiry",
26    }
27
28    for keyword, label in keywords_map.items():
29        if keyword in text_blob and label not in interests:
30            interests.append(label)
31
32    topic = ", ".join(interests) if interests else "general inquiry"
33
34    # Generate summary
35    summary = (
36        f"Telegram conversation about {topic}. "
37        f"User messages: {len(user_messages)}. "
38        f"Last message: '{user_messages[-1][:100]}'"
39    )
40
41    return summary

 1# Proper error handling for CRM integration
 2import asyncio
 3from datetime import datetime
 4
 5async def safe_crm_sync(lead_data: dict, max_retries: int = 3) -> bool:
 6    """
 7    Safe CRM synchronization. On error — retries with exponential backoff. In case of complete failure — saves locally.
 8    """
 9    for attempt in range(max_retries):
10        try:
11            result = await send_to_hubspot(lead_data)
12            if result:
13                print(f"CRM sync success: {lead_data['email']}")
14                return True
15        except Exception as e:
16            wait_time = 2 ** attempt  # 1, 2, 4 seconds
17            print(f"CRM sync attempt {attempt + 1} failed: {e}")
18            print(f"Retrying in {wait_time} seconds...")
19            await asyncio.sleep(wait_time)
20
21    # All attempts failed — saving locally
22    await save_failed_lead(lead_data)
23    print(f"CRM sync FAILED after {max_retries} attempts. Lead saved locally.")
24    return False
25
26async def save_failed_lead(lead_data: dict):
27    """
28    Saves lead locally for retry. In a real project — in Redis or PostgreSQL.
29    """
30    # In production this would be a DB write
31    print(f"Saving failed lead: {lead_data['email']} at {datetime.utcnow()}")