Merge pull request #1064 from airweave-ai/fix/hubspot

fix(hubspot): Use POST search API to avoid 414 URI Too Long errors

Author: orhanrauf

backend/airweave/platform/sources/hubspot.py

@@ -84,6 +84,49 @@ async def _get_with_auth(self, client: httpx.AsyncClient, url: str) -> Dict:
         response.raise_for_status()
         return response.json()
 
+    @retry(
+        stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10), reraise=True
+    )
+    async def _post_with_auth(
+        self, client: httpx.AsyncClient, url: str, json_data: Dict[str, Any]
+    ) -> Dict:
+        """Make authenticated POST request to HubSpot API.
+
+        Args:
+            client: HTTP client
+            url: API endpoint URL
+            json_data: JSON payload for POST body
+
+        Returns:
+            JSON response from API
+        """
+        # Get fresh token (will refresh if needed)
+        access_token = await self.get_access_token()
+        headers = {
+            "Authorization": f"Bearer {access_token}",
+            "Content-Type": "application/json",
+        }
+
+        response = await client.post(url, headers=headers, json=json_data)
+
+        # Handle 401 errors by refreshing token and retrying
+        if response.status_code == 401:
+            self.logger.warning(
+                f"Got 401 Unauthorized from HubSpot API at {url}, refreshing token..."
+            )
+            await self.refresh_on_unauthorized()
+
+            # Get new token and retry
+            access_token = await self.get_access_token()
+            headers = {
+                "Authorization": f"Bearer {access_token}",
+                "Content-Type": "application/json",
+            }
+            response = await client.post(url, headers=headers, json=json_data)
+
+        response.raise_for_status()
+        return response.json()
+
     def _safe_float_conversion(self, value: Any) -> Optional[float]:
         """Safely convert a value to float, handling empty strings and None."""
         if not value or value == "":
@@ -204,16 +247,29 @@ async def _generate_contact_entities(
     ) -> AsyncGenerator[BaseEntity, None]:
         """Generate Contact entities from HubSpot.
 
-        This uses the REST CRM API endpoint for contacts:
-          GET /crm/v3/objects/contacts
+        This uses the POST-based search API to avoid URL length limitations
+        when there are many custom properties:
+          POST /crm/v3/objects/contacts/search
         """
         # Get all available properties for contacts
         all_properties = await self._get_all_properties(client, "contacts")
-        properties_param = ",".join(all_properties)
 
-        url = f"https://api.hubapi.com/crm/v3/objects/contacts?properties={properties_param}"
-        while url:
-            data = await self._get_with_auth(client, url)
+        url = "https://api.hubapi.com/crm/v3/objects/contacts/search"
+        after = None
+        limit = 100
+
+        while True:
+            # Build the search request body with properties in the payload
+            search_body = {
+                "properties": all_properties,
+                "limit": limit,
+            }
+
+            if after:
+                search_body["after"] = after
+
+            data = await self._post_with_auth(client, url, search_body)
+
             for contact in data.get("results", []):
                 raw_properties = contact.get("properties", {})
                 # Clean properties to remove null/empty values
@@ -250,26 +306,41 @@ async def _generate_contact_entities(
                     archived=contact.get("archived", False),
                 )
 
-            # Handle pagination
+            # Handle pagination using 'after' cursor
             paging = data.get("paging", {})
-            next_link = paging.get("next", {}).get("link")
-            url = next_link if next_link else None
+            after = paging.get("next", {}).get("after") if paging else None
+
+            if not after:
+                break
 
     async def _generate_company_entities(
         self, client: httpx.AsyncClient
     ) -> AsyncGenerator[BaseEntity, None]:
         """Generate Company entities from HubSpot.
 
-        This uses the REST CRM API endpoint for companies:
-          GET /crm/v3/objects/companies
+        This uses the POST-based search API to avoid URL length limitations
+        when there are many custom properties:
+          POST /crm/v3/objects/companies/search
         """
         # Get all available properties for companies
         all_properties = await self._get_all_properties(client, "companies")
-        properties_param = ",".join(all_properties)
 
-        url = f"https://api.hubapi.com/crm/v3/objects/companies?properties={properties_param}"
-        while url:
-            data = await self._get_with_auth(client, url)
+        url = "https://api.hubapi.com/crm/v3/objects/companies/search"
+        after = None
+        limit = 100
+
+        while True:
+            # Build the search request body with properties in the payload
+            search_body = {
+                "properties": all_properties,
+                "limit": limit,
+            }
+
+            if after:
+                search_body["after"] = after
+
+            data = await self._post_with_auth(client, url, search_body)
+
             for company in data.get("results", []):
                 raw_properties = company.get("properties", {})
                 # Clean properties to remove null/empty values
@@ -291,25 +362,41 @@ async def _generate_company_entities(
                     archived=company.get("archived", False),
                 )
 
+            # Handle pagination using 'after' cursor
             paging = data.get("paging", {})
-            next_link = paging.get("next", {}).get("link")
-            url = next_link if next_link else None
+            after = paging.get("next", {}).get("after") if paging else None
+
+            if not after:
+                break
 
     async def _generate_deal_entities(
         self, client: httpx.AsyncClient
     ) -> AsyncGenerator[BaseEntity, None]:
         """Generate Deal entities from HubSpot.
 
-        This uses the REST CRM API endpoint for deals:
-          GET /crm/v3/objects/deals
+        This uses the POST-based search API to avoid URL length limitations
+        when there are many custom properties:
+          POST /crm/v3/objects/deals/search
         """
         # Get all available properties for deals
         all_properties = await self._get_all_properties(client, "deals")
-        properties_param = ",".join(all_properties)
 
-        url = f"https://api.hubapi.com/crm/v3/objects/deals?properties={properties_param}"
-        while url:
-            data = await self._get_with_auth(client, url)
+        url = "https://api.hubapi.com/crm/v3/objects/deals/search"
+        after = None
+        limit = 100
+
+        while True:
+            # Build the search request body with properties in the payload
+            search_body = {
+                "properties": all_properties,
+                "limit": limit,
+            }
+
+            if after:
+                search_body["after"] = after
+
+            data = await self._post_with_auth(client, url, search_body)
+
             for deal in data.get("results", []):
                 raw_properties = deal.get("properties", {})
                 # Clean properties to remove null/empty values
@@ -332,25 +419,41 @@ async def _generate_deal_entities(
                     archived=deal.get("archived", False),
                 )
 
+            # Handle pagination using 'after' cursor
             paging = data.get("paging", {})
-            next_link = paging.get("next", {}).get("link")
-            url = next_link if next_link else None
+            after = paging.get("next", {}).get("after") if paging else None
+
+            if not after:
+                break
 
     async def _generate_ticket_entities(
         self, client: httpx.AsyncClient
     ) -> AsyncGenerator[BaseEntity, None]:
         """Generate Ticket entities from HubSpot.
 
-        This uses the REST CRM API endpoint for tickets:
-          GET /crm/v3/objects/tickets
+        This uses the POST-based search API to avoid URL length limitations
+        when there are many custom properties:
+          POST /crm/v3/objects/tickets/search
         """
         # Get all available properties for tickets
         all_properties = await self._get_all_properties(client, "tickets")
-        properties_param = ",".join(all_properties)
 
-        url = f"https://api.hubapi.com/crm/v3/objects/tickets?properties={properties_param}"
-        while url:
-            data = await self._get_with_auth(client, url)
+        url = "https://api.hubapi.com/crm/v3/objects/tickets/search"
+        after = None
+        limit = 100
+
+        while True:
+            # Build the search request body with properties in the payload
+            search_body = {
+                "properties": all_properties,
+                "limit": limit,
+            }
+
+            if after:
+                search_body["after"] = after
+
+            data = await self._post_with_auth(client, url, search_body)
+
             for ticket in data.get("results", []):
                 raw_properties = ticket.get("properties", {})
                 # Clean properties to remove null/empty values
@@ -373,9 +476,12 @@ async def _generate_ticket_entities(
                     archived=ticket.get("archived", False),
                 )
 
+            # Handle pagination using 'after' cursor
             paging = data.get("paging", {})
-            next_link = paging.get("next", {}).get("link")
-            url = next_link if next_link else None
+            after = paging.get("next", {}).get("after") if paging else None
+
+            if not after:
+                break
 
     async def generate_entities(self) -> AsyncGenerator[BaseEntity, None]:
         """Generate all entities from HubSpot.

monke/bongos/hubspot.py

@@ -44,13 +44,15 @@ async def generate_contact_data(token: str):
         async with httpx.AsyncClient(base_url=HUBSPOT_API, timeout=30) as client:
             for token, c in gen_results:
                 await self._pace()
+                # Embed token in multiple fields for better vector search recall
+                # Token in company name is most reliable for vector search
                 payload = {
                     "properties": {
                         "email": c.email,
                         "firstname": c.firstname,
                         "lastname": c.lastname,
+                        "company": f"{c.company or 'Monke Test Corp'} [{token}]",  # Token in company
                         **({"phone": c.phone} if c.phone else {}),
-                        **({"company": c.company} if c.company else {}),
                         **({"address": c.address} if c.address else {}),
                         **({"city": c.city} if c.city else {}),
                         **({"state": c.state} if c.state else {}),
@@ -88,11 +90,15 @@ async def update_entities(self) -> List[Dict[str, Any]]:
         async with httpx.AsyncClient(base_url=HUBSPOT_API, timeout=30) as client:
             for ent in self._contacts[: min(3, len(self._contacts))]:
                 await self._pace()
-                # tiny tweak: add a note-ish detail (mapped to 'company' or 'phone')
+                # Update phone but preserve token in company name
+                token = ent.get("token", "")
                 r = await client.patch(
                     f"/crm/v3/objects/contacts/{ent['id']}",
                     headers=self._hdrs(),
-                    json={"properties": {"company": "Monke QA", "phone": "+1-555-0100"}},
+                    json={"properties": {
+                        "phone": "+1-555-0100",
+                        "company": f"Monke QA Updated [{token}]"  # Preserve token
+                    }},
                 )
                 r.raise_for_status()
                 updated.append({**ent, "updated": True})

monke/configs/hubspot.yaml

@@ -5,11 +5,12 @@ connector:
   type: "hubspot"
   auth_mode: composio
   composio_config:
-    account_id: ca_example_hubspot
-    auth_config_id: ac_example_hubspot
+    account_id: ca_vG0a0ruEje88
+    auth_config_id: ac_lTZKrCIwL1ks
   config_fields:
     openai_model: "gpt-4.1-mini"
     rate_limit_delay_ms: 800
+    post_create_sleep_seconds: 45  # Extra time for HubSpot to index contacts
 test_flow:
   steps: [cleanup, create, sync, verify, update, sync, verify, partial_delete, sync, verify_partial_deletion, verify_remaining_entities, complete_delete, sync, verify_complete_deletion, cleanup]
 entity_count: 3

monke/generation/hubspot.py

@@ -4,22 +4,22 @@
 
 async def generate_hubspot_contact(model: str, token: str) -> HubSpotContact:
     """
-    Generate a realistic CRM contact. The email MUST contain the token (e.g., [email protected])
-    so we can reliably verify later via search.
+    Generate a realistic CRM contact with embedded token for verification.
+    Token will be added to company field for reliable vector search.
     """
     llm = LLMClient(model_override=model)
     instruction = (
         "Generate a realistic CRM contact for a B2B SaaS context. "
-        f"The literal token '{token}' MUST be embedded in the email local-part (e.g., '{token}@example.test') "
-        "and may appear in notes. Include plausible fields."
+        f"The email should be '{token}@monke-test.com'. "
+        "Include realistic firstname, lastname, company name, and other contact details. "
+        "Make it look like a real business contact."
     )
     contact = await llm.generate_structured(HubSpotContact, instruction)
 
     # Ensure invariants
     contact.token = token
     if token not in contact.email:
         # Force tokenized email (stable id)
-        local = f"{token}.contact"
-        contact.email = f"{local}@example.test"
+        contact.email = f"{token}@monke-test.com"
 
     return contact