admin-valentin/arcade-mcp
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Hubspot toolkit (#363)

Browse source 
Reviewer:

1. DM me to get a client ID/secret for the Hubspot app and login
credentials for a sample Hubspot account
2. For now, you'll need to run the engine on the branch
`nate/token-introspection` for the auth flow to work with Hubspot
3. Add the following to `auth.providers` in your engine yaml:

```yaml
- id: arcade-hubspot
  description: 'Hubspot provider'
  enabled: true
  type: oauth2
  provider_id: hubspot
  client_id: ${env:HUBSPOT_CLIENT_ID}
  client_secret: ${env:HUBSPOT_CLIENT_SECRET}
```
This commit is contained in:
Renato Byrro 2025-04-22 17:43:15 -03:00 committed by GitHub
parent 52807bfd0e
commit de227c5f51
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
21 changed files with 1800 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
docker/toolkits.txt
View file

@ -2,10 +2,12 @@ arcade-code-sandbox
arcade-dropbox
arcade-github
arcade-google
arcade-hubspot
arcade-linkedin
arcade-math
arcade-notion-toolkit
arcade-reddit
arcade-salesforce
arcade-search
arcade-slack
arcade-spotify

18
toolkits/hubspot/.pre-commit-config.yaml Normal file
View file

@ -0,0 +1,18 @@
files: ^./
repos:
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: "v4.4.0"
hooks:
- id: check-case-conflict
- id: check-merge-conflict
- id: check-toml
- id: check-yaml
- id: end-of-file-fixer
- id: trailing-whitespace
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.6.7
hooks:
- id: ruff
args: [--fix]
- id: ruff-format

46
toolkits/hubspot/.ruff.toml Normal file
View file

@ -0,0 +1,46 @@
target-version = "py39"
line-length = 100
fix = true
[lint]
select = [
# flake8-2020
"YTT",
# flake8-bandit
"S",
# flake8-bugbear
"B",
# flake8-builtins
"A",
# flake8-comprehensions
"C4",
# flake8-debugger
"T10",
# flake8-simplify
"SIM",
# isort
"I",
# mccabe
"C90",
# pycodestyle
"E", "W",
# pyflakes
"F",
# pygrep-hooks
"PGH",
# pyupgrade
"UP",
# ruff
"RUF",
# tryceratops
"TRY",
]
[lint.per-file-ignores]
"*" = ["TRY003", "B904"]
"**/tests/*" = ["S101", "E501"]
"**/evals/*" = ["S101", "E501"]
[format]
preview = true
skip-magic-trailing-comma = false

21
toolkits/hubspot/LICENSE Normal file
View file

@ -0,0 +1,21 @@
MIT License
Copyright (c) 2025 Arcade
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

53
toolkits/hubspot/Makefile Normal file
View file

@ -0,0 +1,53 @@
.PHONY: help
help:
@echo "🛠️ dropbox Commands:\n"
@grep -E '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) | sort | awk 'BEGIN {FS = ":.*?## "}; {printf "\033[36m%-30s\033[0m %s\n", $$1, $$2}'
.PHONY: install
install: ## Install the poetry environment and install the pre-commit hooks
@echo "📦 Checking if Poetry is installed"
@if ! command -v poetry &> /dev/null; then \
echo "📦 Installing Poetry with pip"; \
pip install poetry==1.8.5; \
else \
echo "📦 Poetry is already installed"; \
fi
@echo "🚀 Installing package in development mode with all extras"
poetry install --all-extras
.PHONY: build
build: clean-build ## Build wheel file using poetry
@echo "🚀 Creating wheel file"
poetry build
.PHONY: clean-build
clean-build: ## clean build artifacts
@echo "🗑️ Cleaning dist directory"
rm -rf dist
.PHONY: test
test: ## Test the code with pytest
@echo "🚀 Testing code: Running pytest"
@poetry run pytest -W ignore -v --cov --cov-config=pyproject.toml --cov-report=xml
.PHONY: coverage
coverage: ## Generate coverage report
@echo "coverage report"
coverage report
@echo "Generating coverage report"
coverage html
.PHONY: bump-version
bump-version: ## Bump the version in the pyproject.toml file
@echo "🚀 Bumping version in pyproject.toml"
poetry version patch
.PHONY: check
check: ## Run code quality tools.
@echo "🚀 Checking Poetry lock file consistency with 'pyproject.toml': Running poetry check"
@poetry check
@echo "🚀 Linting code: Running pre-commit"
@poetry run pre-commit run -a
@echo "🚀 Static type checking: Running mypy"
@poetry run mypy --config-file=pyproject.toml

0
toolkits/hubspot/arcade_hubspot/__init__.py Normal file
View file

17
toolkits/hubspot/arcade_hubspot/constants.py Normal file
View file

@ -0,0 +1,17 @@
import os
HUBSPOT_BASE_URL = "https://api.hubapi.com"
HUBSPOT_CRM_BASE_URL = f"{HUBSPOT_BASE_URL}/crm"
HUBSPOT_DEFAULT_API_VERSION = "v3"
try:
HUBSPOT_MAX_CONCURRENT_REQUESTS = int(os.getenv("HUBSPOT_MAX_CONCURRENT_REQUESTS", 3))
except ValueError:
HUBSPOT_MAX_CONCURRENT_REQUESTS = 3
GLOBALLY_IGNORED_FIELDS = [
"createdate",
"hs_createdate",
"hs_lastmodifieddate",
"lastmodifieddate",
]

9
toolkits/hubspot/arcade_hubspot/custom_critics.py Normal file
View file

@ -0,0 +1,9 @@
from typing import Any
from arcade.sdk.eval import BinaryCritic
class ValueInListCritic(BinaryCritic):
def evaluate(self, expected: list[Any], actual: Any) -> dict[str, float | bool]:
match = actual in expected
return {"match": match, "score": self.weight if match else 0.0}

20
toolkits/hubspot/arcade_hubspot/enums.py Normal file
View file

@ -0,0 +1,20 @@
from enum import Enum
class HubspotObject(Enum):
ACCOUNT = "account"
CALL = "call"
COMMUNICATION = "communication"
COMPANY = "company"
CONTACT = "contact"
DEAL = "deal"
EMAIL = "email"
MEETING = "meeting"
NOTE = "note"
TASK = "task"
@property
def plural(self) -> str:
if self.value == "company":
return "companies"
return f"{self.value}s"

9
toolkits/hubspot/arcade_hubspot/exceptions.py Normal file
View file

@ -0,0 +1,9 @@
from arcade.sdk.errors import ToolExecutionError
class HubspotToolExecutionError(ToolExecutionError):
pass
class NotFoundError(HubspotToolExecutionError):
pass

228
toolkits/hubspot/arcade_hubspot/models.py Normal file
View file

@ -0,0 +1,228 @@
import asyncio
import json
from dataclasses import dataclass
from typing import Any, Optional, cast
import httpx
from arcade_hubspot.constants import (
HUBSPOT_CRM_BASE_URL,
HUBSPOT_DEFAULT_API_VERSION,
HUBSPOT_MAX_CONCURRENT_REQUESTS,
)
from arcade_hubspot.enums import HubspotObject
from arcade_hubspot.exceptions import HubspotToolExecutionError, NotFoundError
from arcade_hubspot.properties import get_object_properties
from arcade_hubspot.utils import clean_data, prepare_api_search_response, remove_none_values
@dataclass
class HubspotCrmClient:
auth_token: str
base_url: str = HUBSPOT_CRM_BASE_URL
max_concurrent_requests: int = HUBSPOT_MAX_CONCURRENT_REQUESTS
_semaphore: asyncio.Semaphore | None = None
def __post_init__(self) -> None:
self._semaphore = self._semaphore or asyncio.Semaphore(self.max_concurrent_requests)
def _raise_for_status(self, response: httpx.Response) -> None:
if response.status_code < 300:
return
try:
data = response.json()
error_message = data["message"]
developer_message = json.dumps(data["errors"])
except Exception:
error_message = response.text
developer_message = None
if response.status_code == 404:
raise NotFoundError(error_message, developer_message)
raise HubspotToolExecutionError(error_message, developer_message)
async def get(
self,
endpoint: str,
params: Optional[dict] = None,
headers: Optional[dict] = None,
api_version: str = HUBSPOT_DEFAULT_API_VERSION,
) -> dict:
headers = headers or {}
headers["Authorization"] = f"Bearer {self.auth_token}"
kwargs = {
"url": f"{self.base_url}/{api_version}/{endpoint}",
"headers": headers,
}
if isinstance(params, dict):
kwargs["params"] = params
async with self._semaphore, httpx.AsyncClient() as client: # type: ignore[union-attr]
response = await client.get(**kwargs) # type: ignore[arg-type]
self._raise_for_status(response)
return cast(dict, response.json())
async def post(
self,
endpoint: str,
data: Optional[dict] = None,
json_data: Optional[dict] = None,
headers: Optional[dict] = None,
api_version: str = HUBSPOT_DEFAULT_API_VERSION,
) -> dict:
headers = headers or {}
headers["Authorization"] = f"Bearer {self.auth_token}"
headers["Content-Type"] = "application/json"
kwargs = {
"url": f"{self.base_url}/{api_version}/{endpoint}",
"headers": headers,
}
if data and json_data:
raise ValueError("Cannot provide both data and json_data")
if data:
kwargs["data"] = data
elif json_data:
kwargs["json"] = json_data
async with self._semaphore, httpx.AsyncClient() as client: # type: ignore[union-attr]
response = await client.post(**kwargs) # type: ignore[arg-type]
self._raise_for_status(response)
return cast(dict, response.json())
async def get_associated_objects(
self,
parent_object: HubspotObject,
parent_id: str,
associated_object: HubspotObject,
limit: int = 10,
after: Optional[str] = None,
properties: Optional[list[str]] = None,
) -> list[dict]:
endpoint = (
f"objects/{parent_object.value}/{parent_id}/associations/{associated_object.value}"
)
params = {
"limit": limit,
}
if after:
params["after"] = after # type: ignore[assignment]
response = await self.get(endpoint, params=params, api_version="v4")
if not response["results"]:
return []
return await self.batch_get_objects(
object_type=associated_object,
object_ids=[object_data["toObjectId"] for object_data in response["results"]],
properties=properties or get_object_properties(associated_object),
)
async def get_object_by_id(
self,
object_type: HubspotObject,
object_id: str,
properties: Optional[list[str]] = None,
) -> dict:
endpoint = f"objects/{object_type.plural}/{object_id}"
params = {}
if properties:
params["properties"] = properties
return clean_data(await self.get(endpoint, params=params), object_type)
async def batch_get_objects(
self,
object_type: HubspotObject,
object_ids: list[str],
properties: Optional[list[str]] = None,
) -> list[dict]:
endpoint = f"objects/{object_type.plural}/batch/read"
data: dict[str, Any] = {"inputs": [{"id": object_id} for object_id in object_ids]}
if properties:
data["properties"] = properties
response = await self.post(endpoint, json_data=data)
return [clean_data(object_data, object_type) for object_data in response["results"]]
async def search_by_keywords(
self,
object_type: HubspotObject,
keywords: str,
limit: int = 10,
next_page_token: Optional[str] = None,
associations: Optional[list[HubspotObject]] = None,
) -> dict:
if not keywords:
raise HubspotToolExecutionError("`keywords` must be a non-empty string")
associations = associations or []
endpoint = f"objects/{object_type.plural}/search"
request_data = {
"query": keywords,
"limit": limit,
"sorts": [{"propertyName": "hs_lastmodifieddate", "direction": "DESCENDING"}],
"properties": get_object_properties(object_type),
}
if next_page_token:
request_data["after"] = next_page_token
data = prepare_api_search_response(
data=await self.post(endpoint, json_data=request_data),
object_type=object_type,
)
for object_ in data[object_type.plural]:
for association in associations:
results = await self.get_associated_objects(
parent_object=object_type,
parent_id=object_["id"],
associated_object=association,
limit=10,
)
if results:
object_[association.plural] = results
return data
async def create_contact(
self,
company_id: str,
first_name: str,
last_name: Optional[str] = None,
email: Optional[str] = None,
phone: Optional[str] = None,
mobile_phone: Optional[str] = None,
job_title: Optional[str] = None,
) -> dict:
request_data = {
"associations": [
{
"types": [
{
"associationCategory": "HUBSPOT_DEFINED",
"associationTypeId": "1",
}
],
"to": {"id": company_id},
},
],
"properties": remove_none_values({
"firstname": first_name,
"lastname": last_name,
"email": email,
"phone": phone,
"mobilephone": mobile_phone,
"jobtitle": job_title,
}),
}
endpoint = "objects/contacts"
return await self.post(endpoint, json_data=request_data)

127
toolkits/hubspot/arcade_hubspot/properties.py Normal file
View file

@ -0,0 +1,127 @@
from arcade_hubspot.enums import HubspotObject
HUBSPOT_OBJECT_PROPERTIES = {
HubspotObject.COMPANY: [
"type",
"name",
"about_us",
"address",
"city",
"state",
"zip",
"country",
"annualrevenue",
"hs_annual_revenue_currency_code",
"industry",
"phone",
"website",
"domain",
"numberofemployees",
"hs_lead_status",
"lifecyclestage",
"linkedin_company_page",
"twitterhandle",
],
HubspotObject.CONTACT: [
"salutation",
"email",
"work_email",
"hs_additional_emails",
"mobilephone",
"phone",
"firstname",
"lastname",
"lifecyclestage",
"annualrevenue",
"address",
"date_of_birth",
"degree",
"gender",
"buying_role",
"hs_language",
"hs_lead_status",
"hs_timezone",
"hubspot_owner_id",
"hubspot_owner_name",
"job_function",
"jobtitle",
"seniority",
"industry",
"twitterhandle",
],
HubspotObject.CALL: [
"hs_body_preview",
"hs_call_direction",
"hs_call_status",
"hs_call_summary",
"hs_call_title",
"hs_createdate",
"hubspot_owner_id",
"hs_call_disposition",
"hs_timestamp",
],
HubspotObject.DEAL: [
"dealname",
"dealstage",
"dealtype",
"closedate",
"closed_lost_reason",
"closed_won_reason",
"hs_is_closed_won",
"hs_is_closed_lost",
"hs_closed_amount",
"pipeline",
"hubspot_owner_id",
"description",
"hs_deal_score",
"amount",
"hs_forecast_probability",
"hs_forecast_amount",
],
HubspotObject.EMAIL: [
"hs_object_id",
"hs_body_preview",
"hs_email_sender_raw",
"hs_email_from_raw",
"hs_email_to_raw",
"hs_email_bcc_raw",
"hs_email_cc_raw",
"hs_email_subject",
"hs_email_status",
"hs_timestamp",
"hubspot_owner_id",
"hs_email_associated_contact_id",
],
HubspotObject.MEETING: [
"hs_object_id",
"hs_meeting_title",
"hs_body_preview",
"hs_meeting_location",
"hs_meeting_outcome",
"hubspot_owner_id",
"hs_meeting_start_time",
"hs_meeting_end_time",
],
HubspotObject.NOTE: [
"hs_object_id",
"hs_body_preview",
"hs_timestamp",
"hubspot_owner_id",
],
HubspotObject.TASK: [
"hs_object_id",
"hs_body_preview",
"hs_timestamp",
"hubspot_owner_id",
"hs_associated_contact_labels",
"hs_task_is_overdue",
"hs_task_priority",
"hs_task_status",
"hs_task_subject",
"hs_task_type",
],
}
def get_object_properties(object_type: HubspotObject) -> list[str]:
return HUBSPOT_OBJECT_PROPERTIES[object_type]

8
toolkits/hubspot/arcade_hubspot/tools/__init__.py Normal file
View file

@ -0,0 +1,8 @@
from arcade_hubspot.tools.crm.companies import get_company_data_by_keywords
from arcade_hubspot.tools.crm.contacts import create_contact, get_contact_data_by_keywords
__all__ = [
"get_company_data_by_keywords",
"get_contact_data_by_keywords",
"create_contact",
]

63
toolkits/hubspot/arcade_hubspot/tools/crm/companies.py Normal file
View file

@ -0,0 +1,63 @@
from typing import Annotated, Any, Optional
from arcade.sdk import ToolContext, tool
from arcade.sdk.auth import OAuth2
from arcade_hubspot.enums import HubspotObject
from arcade_hubspot.models import HubspotCrmClient
@tool(
requires_auth=OAuth2(
id="hubspot",
scopes=[
"oauth",
"crm.objects.companies.read",
"crm.objects.contacts.read",
"crm.objects.deals.read",
"sales-email-read",
],
),
)
async def get_company_data_by_keywords(
context: ToolContext,
keywords: Annotated[
str,
"The keywords to search for companies. It will match against the company name, phone, "
"and website.",
],
limit: Annotated[
int, "The maximum number of companies to return. Defaults to 10. Max is 10."
] = 10,
next_page_token: Annotated[
Optional[str],
"The token to get the next page of results. "
"Defaults to None (returns first page of results)",
] = None,
) -> Annotated[
dict[str, Any],
"Retrieve company data with associated contacts, deals, calls, emails, "
"meetings, notes, and tasks.",
]:
"""Retrieve company data with associated contacts, deals, calls, emails,
meetings, notes, and tasks.
This tool will return up to 10 items of each associated object (contacts, leads, etc).
"""
limit = min(limit, 10)
client = HubspotCrmClient(context.get_auth_token_or_empty())
return await client.search_by_keywords(
object_type=HubspotObject.COMPANY,
keywords=keywords,
limit=limit,
next_page_token=next_page_token,
associations=[
HubspotObject.CALL,
HubspotObject.CONTACT,
HubspotObject.DEAL,
HubspotObject.EMAIL,
HubspotObject.MEETING,
HubspotObject.NOTE,
HubspotObject.TASK,
],
)

91
toolkits/hubspot/arcade_hubspot/tools/crm/contacts.py Normal file
View file

@ -0,0 +1,91 @@
from typing import Annotated, Any, Optional
from arcade.sdk import ToolContext, tool
from arcade.sdk.auth import OAuth2
from arcade_hubspot.enums import HubspotObject
from arcade_hubspot.models import HubspotCrmClient
from arcade_hubspot.utils import clean_data
@tool(
requires_auth=OAuth2(
id="hubspot",
scopes=[
"oauth",
"crm.objects.contacts.read",
],
),
)
async def get_contact_data_by_keywords(
context: ToolContext,
keywords: Annotated[
str,
"The keywords to search for contacts. It will match against the contact's "
"first and last name, email addresses, phone numbers, and company name.",
],
limit: Annotated[
int, "The maximum number of contacts to return. Defaults to 10. Max is 100."
] = 10,
next_page_token: Annotated[
Optional[str],
"The token to get the next page of results. "
"Defaults to None (returns first page of results)",
] = None,
) -> Annotated[
dict[str, Any],
"Retrieve contact data with associated companies, deals, calls, "
"emails, meetings, notes, and tasks.",
]:
"""
Retrieve contact data with associated companies, deals, calls, emails,
meetings, notes, and tasks.
"""
limit = min(limit, 100)
client = HubspotCrmClient(context.get_auth_token_or_empty())
return await client.search_by_keywords(
object_type=HubspotObject.CONTACT,
keywords=keywords,
limit=limit,
next_page_token=next_page_token,
associations=[
HubspotObject.CALL,
HubspotObject.COMPANY,
HubspotObject.DEAL,
HubspotObject.EMAIL,
HubspotObject.MEETING,
HubspotObject.NOTE,
HubspotObject.TASK,
],
)
@tool(
requires_auth=OAuth2(
id="hubspot",
scopes=["oauth", "crm.objects.contacts.write"],
),
)
async def create_contact(
context: ToolContext,
company_id: Annotated[str, "The ID of the company to create the contact for."],
first_name: Annotated[str, "The first name of the contact."],
last_name: Annotated[Optional[str], "The last name of the contact."] = None,
email: Annotated[Optional[str], "The email address of the contact."] = None,
phone: Annotated[Optional[str], "The phone number of the contact."] = None,
mobile_phone: Annotated[Optional[str], "The mobile phone number of the contact."] = None,
job_title: Annotated[Optional[str], "The job title of the contact."] = None,
) -> Annotated[dict, "Create a contact associated with a company."]:
"""Create a contact associated with a company."""
client = HubspotCrmClient(context.get_auth_token_or_empty())
response = await client.create_contact(
company_id=company_id,
first_name=first_name,
last_name=last_name,
email=email,
phone=phone,
mobile_phone=mobile_phone,
job_title=job_title,
)
return clean_data(response, HubspotObject.CONTACT)

313
toolkits/hubspot/arcade_hubspot/utils.py Normal file
View file

@ -0,0 +1,313 @@
from typing import Any, Callable
from arcade_hubspot.constants import GLOBALLY_IGNORED_FIELDS
from arcade_hubspot.enums import HubspotObject
def remove_none_values(data: dict) -> dict:
cleaned: dict[str, Any] = {}
for key, value in data.items():
if value is None or key in GLOBALLY_IGNORED_FIELDS:
continue
if isinstance(value, dict):
cleaned_dict = remove_none_values(value)
if cleaned_dict:
cleaned[key] = cleaned_dict
elif isinstance(value, (list, tuple, set)):
collection_type = type(value)
cleaned_list = [remove_none_values(item) for item in value]
if cleaned_list:
cleaned[key] = collection_type(cleaned_list)
else:
cleaned[key] = value
return cleaned
def prepare_api_search_response(data: dict, object_type: HubspotObject) -> dict:
response: dict[str, Any] = {
object_type.plural: [clean_data(company, object_type) for company in data["results"]],
}
after = data.get("paging", {}).get("next", {}).get("after")
if after:
response["pagination"] = {
"are_there_more_results?": True,
"next_page_token": after,
}
else:
response["pagination"] = {
"are_there_more_results?": False,
}
return response
def rename_dict_keys(data: dict, rename: dict) -> dict:
for old_key, new_key in rename.items():
if old_key in data:
data[new_key] = data[old_key]
data.pop(old_key, None)
return data
def global_cleaner(clean_func: Callable[[dict], dict]) -> Callable[[dict], dict]:
def global_cleaner(data: dict) -> dict:
cleaned_data: dict[str, Any] = {}
if "hs_object_id" in data:
cleaned_data["id"] = data["hs_object_id"]
del data["hs_object_id"]
data = rename_dict_keys(
data,
{
"hubspot_owner_id": "owner_id",
"hs_timestamp": "datetime",
"hs_body_preview": "body",
},
)
for key, value in data.items():
if key in GLOBALLY_IGNORED_FIELDS or value is None:
continue
if isinstance(value, dict):
cleaned_data[key] = global_cleaner(value)
elif isinstance(value, (list, tuple, set)):
cleaned_items = []
for item in value:
if isinstance(item, dict):
cleaned_items.append(global_cleaner(item))
else:
cleaned_items.append(item)
cleaned_data[key] = cleaned_items
else:
cleaned_data[key] = value
return cleaned_data
def wrapper(data: dict) -> dict:
return remove_none_values(global_cleaner(clean_func(data["properties"])))
return wrapper
def clean_data(data: dict, object_type: HubspotObject) -> dict:
_mapping = {
HubspotObject.CALL: clean_call_data,
HubspotObject.COMPANY: clean_company_data,
HubspotObject.CONTACT: clean_contact_data,
HubspotObject.DEAL: clean_deal_data,
HubspotObject.EMAIL: clean_email_data,
HubspotObject.MEETING: clean_meeting_data,
HubspotObject.NOTE: clean_note_data,
HubspotObject.TASK: clean_task_data,
}
try:
return _mapping[object_type](data)
except KeyError:
raise ValueError(f"Unsupported object type: {object_type}")
@global_cleaner
def clean_company_data(data: dict) -> dict:
data["object_type"] = HubspotObject.COMPANY.value
data["website"] = data.get("website", data.get("domain"))
data.pop("domain", None)
data["address"] = {
"street": data.get("address"),
"city": data.get("city"),
"state": data.get("state"),
"zip": data.get("zip"),
"country": data.get("country"),
}
data.pop("address", None)
data.pop("city", None)
data.pop("state", None)
data.pop("zip", None)
data.pop("country", None)
data["annual_revenue"] = {
"value": data.get("annualrevenue"),
"currency": data.get("hs_annual_revenue_currency_code"),
}
data.pop("annualrevenue", None)
data.pop("hs_annual_revenue_currency_code", None)
rename = {
"linkedin_company_page": "linkedin_url",
"numberofemployees": "employee_count",
"type": "company_type",
"annualrevenue": "annual_revenue",
"lifecyclestage": "lifecycle_stage",
"hs_lead_status": "lead_status",
}
data = rename_dict_keys(data, rename)
return data
@global_cleaner
def clean_contact_data(data: dict) -> dict:
data["object_type"] = HubspotObject.CONTACT.value
rename = {
"lifecyclestage": "lifecycle_stage",
"hs_lead_status": "lead_status",
"email": "email_address",
}
data = rename_dict_keys(data, rename)
return data
@global_cleaner
def clean_deal_data(data: dict) -> dict:
data["object_type"] = HubspotObject.DEAL.value
if data.get("closedate") or data.get("hs_closed_amount"):
data["close"] = {
"is_closed": data.get("hs_is_closed"),
"date": data.get("closedate"),
"amount": data.get("hs_closed_amount"),
}
if data.get("hs_is_closed_won") in ["true", True]:
data["close"]["status"] = "won"
data["close"]["status_reason"] = data.get("closed_won_reason")
elif data.get("hs_is_closed_lost") in ["true", True]:
data["close"]["status"] = "lost"
data["close"]["status_reason"] = data.get("closed_lost_reason")
if data.get("amount"):
data["amount"] = {
"value": data["amount"],
"currency": data.get("deal_currency_code"),
}
if data.get("hs_forecast_probability"):
data["amount"]["forecast"] = {
"probability": data["hs_forecast_probability"],
"expected_value": data.get("hs_forecast_amount"),
}
rename = {
"dealname": "name",
"dealstage": "stage",
"dealscore": "score",
"dealtype": "type",
}
data = rename_dict_keys(data, rename)
data.pop("hs_is_closed", None)
data.pop("closedate", None)
data.pop("hs_closed_amount", None)
data.pop("deal_currency_code", None)
data.pop("close_won_reason", None)
data.pop("close_lost_reason", None)
data.pop("hs_is_closed_won", None)
data.pop("hs_is_closed_lost", None)
data.pop("hs_forecast_probability", None)
data.pop("hs_forecast_amount", None)
return data
@global_cleaner
def clean_email_data(data: dict) -> dict:
data["object_type"] = HubspotObject.EMAIL.value
rename = {
"hs_body_preview": "body",
"hs_email_from_raw": "from",
"hs_email_to_raw": "to",
"hs_email_bcc_raw": "bcc",
"hs_email_cc_raw": "cc",
"hs_email_subject": "subject",
"hs_email_status": "status",
"hs_email_associated_contact_id": "contact_id",
}
data = rename_dict_keys(data, rename)
return data
@global_cleaner
def clean_call_data(data: dict) -> dict:
data["object_type"] = HubspotObject.CALL.value
rename = {
"hs_call_direction": "direction",
"hs_call_status": "status",
"hs_call_summary": "summary",
"hs_call_title": "title",
"hs_call_disposition": "outcome",
}
data = rename_dict_keys(data, rename)
return data
@global_cleaner
def clean_note_data(data: dict) -> dict:
data["object_type"] = HubspotObject.NOTE.value
return data
@global_cleaner
def clean_meeting_data(data: dict) -> dict:
data["object_type"] = HubspotObject.MEETING.value
rename = {
"hs_meeting_outcome": "outcome",
"hs_meeting_location": "location",
"hs_meeting_start_time": "start_time",
"hs_meeting_end_time": "end_time",
}
data = rename_dict_keys(data, rename)
data["content"] = {
"title": data.get("hs_meeting_title"),
"body": data.get("hs_body_preview"),
}
data.pop("hs_meeting_title", None)
data.pop("hs_body_preview", None)
data["schedule"] = {
"start_time": data.get("start_time"),
"end_time": data.get("end_time"),
}
data.pop("start_time", None)
data.pop("end_time", None)
return data
@global_cleaner
def clean_task_data(data: dict) -> dict:
data["object_type"] = HubspotObject.TASK.value
if data.get("hs_task_priority") == "NONE":
data["hs_task_priority"] = None
rename = {
"hs_task_status": "status",
"hs_task_priority": "priority",
"hs_task_missed_due_date": "missed_due_date",
"hs_task_type": "task_type",
"hs_associated_contact_labels": "associated_contact",
}
data = rename_dict_keys(data, rename)
data["content"] = {
"body": data.get("hs_body_preview"),
"subject": data.get("hs_task_subject"),
}
data.pop("hs_body_preview", None)
data.pop("hs_task_subject", None)
data["schedule"] = {
"datetime": data.get("hs_timestamp"),
"is_overdue": data.get("hs_task_is_overdue"),
}
data.pop("hs_timestamp", None)
data.pop("hs_task_is_overdue", None)
return data

16
toolkits/hubspot/conftest.py Normal file
View file

@ -0,0 +1,16 @@
from unittest.mock import patch
import pytest
from arcade.sdk import ToolAuthorizationContext, ToolContext
@pytest.fixture
def mock_context():
mock_auth = ToolAuthorizationContext(token="fake-token") # noqa: S106
return ToolContext(authorization=mock_auth)
@pytest.fixture
def mock_httpx_client():
with patch("arcade_hubspot.models.httpx") as mock_httpx:
yield mock_httpx.AsyncClient().__aenter__.return_value

194
toolkits/hubspot/evals/eval_crm_companies.py Normal file
View file

@ -0,0 +1,194 @@
from arcade.sdk import ToolCatalog
from arcade.sdk.eval import (
BinaryCritic,
EvalRubric,
EvalSuite,
ExpectedToolCall,
tool_eval,
)
import arcade_hubspot
from arcade_hubspot.tools import get_company_data_by_keywords
rubric = EvalRubric(
fail_threshold=0.8,
warn_threshold=0.9,
)
catalog = ToolCatalog()
catalog.add_module(arcade_hubspot)
@tool_eval()
def get_company_data_by_keywords_eval_suite() -> EvalSuite:
"""Create an evaluation suite for the get_company_data_by_keywords tool."""
suite = EvalSuite(
name="Get Company Data by Keywords",
system_message="You are an AI assistant that can interact with Hubspot CRM using the provided tools.",
catalog=catalog,
rubric=rubric,
)
suite.add_case(
name="Get company data by keywords",
user_message="Get information about the Acme company.",
expected_tool_calls=[
ExpectedToolCall(
func=get_company_data_by_keywords,
args={
"keywords": "Acme",
"limit": 10,
"next_page_token": None,
},
),
],
rubric=rubric,
critics=[
BinaryCritic(critic_field="keywords", weight=0.8),
BinaryCritic(critic_field="next_page_token", weight=0.2),
],
)
suite.add_case(
name="Get company data by keywords with limit",
user_message="Get information of up to 3 companies with the word 'Acme' in their name.",
expected_tool_calls=[
ExpectedToolCall(
func=get_company_data_by_keywords,
args={
"keywords": "Acme",
"limit": 3,
"next_page_token": None,
},
),
],
rubric=rubric,
critics=[
BinaryCritic(critic_field="keywords", weight=0.6),
BinaryCritic(critic_field="limit", weight=0.3),
BinaryCritic(critic_field="next_page_token", weight=0.1),
],
)
suite.add_case(
name="Get summary of company activity",
user_message="Get a summary of the latest activities in the Acme company.",
expected_tool_calls=[
ExpectedToolCall(
func=get_company_data_by_keywords,
args={
"keywords": "Acme",
"limit": 10,
"next_page_token": None,
},
),
],
rubric=rubric,
critics=[
BinaryCritic(critic_field="keywords", weight=0.8),
BinaryCritic(critic_field="next_page_token", weight=0.2),
],
)
suite.add_case(
name="Interactions with contacts of an account",
user_message="Get me the interactions with the contacts of the Acme company.",
expected_tool_calls=[
ExpectedToolCall(
func=get_company_data_by_keywords,
args={
"keywords": "Acme",
"limit": 10,
"next_page_token": None,
},
),
],
rubric=rubric,
critics=[
BinaryCritic(critic_field="keywords", weight=0.8),
BinaryCritic(critic_field="next_page_token", weight=0.2),
],
)
suite.add_case(
name="Emails or calls with contacts of an account",
user_message="Were there any emails or calls with contacts of the Acme company this week?",
expected_tool_calls=[
ExpectedToolCall(
func=get_company_data_by_keywords,
args={
"keywords": "Acme",
"limit": 10,
"next_page_token": None,
},
),
],
rubric=rubric,
critics=[
BinaryCritic(critic_field="keywords", weight=0.8),
BinaryCritic(critic_field="next_page_token", weight=0.2),
],
)
suite.add_case(
name="Get company status",
user_message="What's the status of the Acme company?",
expected_tool_calls=[
ExpectedToolCall(
func=get_company_data_by_keywords,
args={
"keywords": "Acme",
"limit": 10,
"next_page_token": None,
},
),
],
rubric=rubric,
critics=[
BinaryCritic(critic_field="keywords", weight=0.8),
BinaryCritic(critic_field="next_page_token", weight=0.2),
],
)
suite.add_case(
name="Get overdue tasks",
user_message="Are there any tasks overdue for the Acme company?",
expected_tool_calls=[
ExpectedToolCall(
func=get_company_data_by_keywords,
args={
"keywords": "Acme",
"limit": 10,
"next_page_token": None,
},
),
],
rubric=rubric,
critics=[
BinaryCritic(critic_field="keywords", weight=0.8),
BinaryCritic(critic_field="next_page_token", weight=0.2),
],
)
suite.add_case(
name="Get company closing likelihood",
user_message="What's the likelihood of the Acme company closing a deal?",
expected_tool_calls=[
ExpectedToolCall(
func=get_company_data_by_keywords,
args={
"keywords": "Acme",
"limit": 10,
"next_page_token": None,
},
),
],
rubric=rubric,
critics=[
BinaryCritic(critic_field="keywords", weight=0.8),
BinaryCritic(critic_field="next_page_token", weight=0.2),
],
)
return suite

197
toolkits/hubspot/evals/eval_crm_contacts.py Normal file
View file

@ -0,0 +1,197 @@
from arcade.sdk import ToolCatalog
from arcade.sdk.eval import (
BinaryCritic,
EvalRubric,
EvalSuite,
ExpectedToolCall,
tool_eval,
)
import arcade_hubspot
from arcade_hubspot.custom_critics import ValueInListCritic
from arcade_hubspot.tools import create_contact, get_contact_data_by_keywords
rubric = EvalRubric(
fail_threshold=0.8,
warn_threshold=0.9,
)
catalog = ToolCatalog()
catalog.add_module(arcade_hubspot)
@tool_eval()
def get_contact_data_by_keywords_eval_suite() -> EvalSuite:
"""Create an evaluation suite for the get_contact_data_by_keywords tool."""
suite = EvalSuite(
name="Get Contact Data by Keywords",
system_message="You are an AI assistant that can interact with Hubspot CRM using the provided tools.",
catalog=catalog,
rubric=rubric,
)
suite.add_case(
name="Get contact data by keywords",
user_message="Get information about the Jason Bourne contact.",
expected_tool_calls=[
ExpectedToolCall(
func=get_contact_data_by_keywords,
args={
"keywords": "Jason Bourne",
"limit": 10,
"next_page_token": None,
},
),
],
rubric=rubric,
critics=[
BinaryCritic(critic_field="keywords", weight=0.8),
BinaryCritic(critic_field="next_page_token", weight=0.2),
],
)
suite.add_case(
name="Get contact data by keywords with limit",
user_message="Get information of up to 3 contacts with last name 'Bourne'.",
expected_tool_calls=[
ExpectedToolCall(
func=get_contact_data_by_keywords,
args={
"keywords": "Bourne",
"limit": 3,
"next_page_token": None,
},
),
],
rubric=rubric,
critics=[
BinaryCritic(critic_field="keywords", weight=0.7),
BinaryCritic(critic_field="limit", weight=0.15),
BinaryCritic(critic_field="next_page_token", weight=0.15),
],
)
suite.add_case(
name="Get summary of contact activity",
user_message="Get a summary of the latest activities with the Jason Bourne contact.",
expected_tool_calls=[
ExpectedToolCall(
func=get_contact_data_by_keywords,
args={
"keywords": "Jason Bourne",
"limit": 10,
"next_page_token": None,
},
),
],
rubric=rubric,
critics=[
BinaryCritic(critic_field="keywords", weight=0.8),
BinaryCritic(critic_field="next_page_token", weight=0.2),
],
)
suite.add_case(
name="Interactions with contacts of an account",
user_message="Get me the interactions with the Jason Bourne contact from Acme.",
expected_tool_calls=[
ExpectedToolCall(
func=get_contact_data_by_keywords,
args={
"keywords": ["Jason Bourne", "Jason Bourne Acme"],
"limit": 10,
"next_page_token": None,
},
),
],
rubric=rubric,
critics=[
ValueInListCritic(critic_field="keywords", weight=0.8),
BinaryCritic(critic_field="next_page_token", weight=0.2),
],
)
suite.add_case(
name="Emails or calls with contacts of an account",
user_message="Were there any emails or calls with the Jason Bourne contact from Acme this week?",
expected_tool_calls=[
ExpectedToolCall(
func=get_contact_data_by_keywords,
args={
"keywords": ["Jason Bourne", "Jason Bourne Acme"],
"limit": 10,
"next_page_token": None,
},
),
],
rubric=rubric,
critics=[
ValueInListCritic(critic_field="keywords", weight=0.8),
BinaryCritic(critic_field="next_page_token", weight=0.2),
],
)
suite.add_case(
name="Get overdue tasks",
user_message="Are there any tasks overdue for the Jason Bourne contact from Acme?",
expected_tool_calls=[
ExpectedToolCall(
func=get_contact_data_by_keywords,
args={
"keywords": ["Jason Bourne", "Jason Bourne Acme"],
"limit": 10,
"next_page_token": None,
},
),
],
rubric=rubric,
critics=[
ValueInListCritic(critic_field="keywords", weight=0.8),
BinaryCritic(critic_field="next_page_token", weight=0.2),
],
)
return suite
@tool_eval()
def create_contact_eval_suite() -> EvalSuite:
"""Create an evaluation suite for the create_contact tool."""
suite = EvalSuite(
name="Create Contact",
system_message="You are an AI assistant that can interact with Hubspot CRM using the provided tools.",
catalog=catalog,
rubric=rubric,
)
suite.add_case(
name="Create contact",
user_message="Create a contact with the following information: first name 'Jason', "
"last name 'Bourne', email 'jason.bourne@acme.com', phone '+1234567890', "
"and job title 'Unbeatable', and associated with company id '1234567890'",
expected_tool_calls=[
ExpectedToolCall(
func=create_contact,
args={
"company_id": "1234567890",
"first_name": "Jason",
"last_name": "Bourne",
"email": "jason.bourne@acme.com",
"phone": "+1234567890",
"job_title": "Unbeatable",
},
),
],
rubric=rubric,
critics=[
BinaryCritic(critic_field="company_id", weight=1 / 6),
BinaryCritic(critic_field="first_name", weight=1 / 6),
BinaryCritic(critic_field="last_name", weight=1 / 6),
BinaryCritic(critic_field="email", weight=1 / 6),
BinaryCritic(critic_field="phone", weight=1 / 6),
BinaryCritic(critic_field="job_title", weight=1 / 6),
],
)
return suite

42
toolkits/hubspot/pyproject.toml Normal file
View file

@ -0,0 +1,42 @@
[tool.poetry]
name = "arcade_hubspot"
version = "0.1.0"
description = "Arcade tools designed for LLMs to interact with Hubspot"
authors = ["Arcade <dev@arcade.dev>"]
[tool.poetry.dependencies]
python = "^3.10"
arcade-ai = ">=1.0.5,<2.0"
httpx = "^0.27.2"
[tool.poetry.dev-dependencies]
pytest = "^8.3.0"
pytest-cov = "^4.0.0"
pytest-asyncio = "^0.24.0"
pytest-mock = "^3.11.1"
mypy = "^1.5.1"
pre-commit = "^3.4.0"
tox = "^4.11.1"
ruff = "^0.7.4"
[build-system]
requires = ["poetry-core>=1.0.0"]
build-backend = "poetry.core.masonry.api"
[tool.mypy]
files = ["arcade_hubspot/**/*.py"]
python_version = "3.10"
disallow_untyped_defs = "True"
disallow_any_unimported = "True"
no_implicit_optional = "True"
check_untyped_defs = "True"
warn_return_any = "True"
warn_unused_ignores = "True"
show_error_codes = "True"
ignore_missing_imports = "True"
[tool.pytest.ini_options]
testpaths = ["tests"]
[tool.coverage.report]
skip_empty = true

326
toolkits/hubspot/tests/test_hubspot_client.py Normal file
View file

@ -0,0 +1,326 @@
from copy import deepcopy
from unittest.mock import AsyncMock, MagicMock, call, patch
import httpx
import pytest
from arcade_hubspot.enums import HubspotObject
from arcade_hubspot.exceptions import HubspotToolExecutionError, NotFoundError
from arcade_hubspot.models import HubspotCrmClient
@pytest.mark.asyncio
async def test_get_success(mock_context, mock_httpx_client):
client = HubspotCrmClient(mock_context.get_auth_token_or_empty())
mock_response = MagicMock(spec=httpx.Response)
mock_response.status_code = 200
mock_response.json.return_value = {"data": []}
mock_httpx_client.get.return_value = mock_response
response = await client.get("objects/contacts", params={"id": "123"})
assert response == {"data": []}
mock_httpx_client.get.assert_called_once_with(
url="https://api.hubapi.com/crm/v3/objects/contacts",
params={"id": "123"},
headers={"Authorization": f"Bearer {mock_context.get_auth_token_or_empty()}"},
)
@pytest.mark.asyncio
async def test_get_not_found_error(mock_context, mock_httpx_client):
client = HubspotCrmClient(mock_context.get_auth_token_or_empty())
mock_response = MagicMock(spec=httpx.Response)
mock_response.status_code = 404
mock_response.json.return_value = {"message": "Not Found", "errors": [{"message": "Not Found"}]}
mock_httpx_client.get.return_value = mock_response
with pytest.raises(NotFoundError) as exc_info:
await client.get("objects/contacts", params={"id": "123"})
assert exc_info.value.message == "Not Found"
assert exc_info.value.developer_message == '[{"message": "Not Found"}]'
@pytest.mark.asyncio
async def test_get_internal_server_error(mock_context, mock_httpx_client):
client = HubspotCrmClient(mock_context.get_auth_token_or_empty())
mock_response = MagicMock(spec=httpx.Response)
mock_response.status_code = 500
mock_response.json.return_value = {
"message": "Internal Server Error",
"errors": [{"message": "Internal Server Error"}],
}
mock_httpx_client.get.return_value = mock_response
with pytest.raises(HubspotToolExecutionError) as exc_info:
await client.get("objects/contacts", params={"id": "123"})
assert exc_info.value.message == "Internal Server Error"
assert exc_info.value.developer_message == '[{"message": "Internal Server Error"}]'
@pytest.mark.asyncio
async def test_get_with_invalid_json_error_response(mock_context, mock_httpx_client):
client = HubspotCrmClient(mock_context.get_auth_token_or_empty())
mock_response = MagicMock(spec=httpx.Response)
mock_response.status_code = 500
mock_response.text = "Text error message"
mock_response.json.side_effect = ValueError("Invalid JSON")
mock_httpx_client.get.return_value = mock_response
with pytest.raises(HubspotToolExecutionError) as exc_info:
await client.get("objects/contacts", params={"id": "123"})
assert exc_info.value.message == "Text error message"
assert exc_info.value.developer_message is None
@pytest.mark.asyncio
async def test_post_success(mock_context, mock_httpx_client):
client = HubspotCrmClient(mock_context.get_auth_token_or_empty())
mock_response = MagicMock(spec=httpx.Response)
mock_response.status_code = 200
mock_response.json.return_value = {"data": []}
mock_httpx_client.post.return_value = mock_response
response = await client.post("objects/contacts", json_data={"id": "123"})
assert response == {"data": []}
mock_httpx_client.post.assert_called_once_with(
url="https://api.hubapi.com/crm/v3/objects/contacts",
json={"id": "123"},
headers={
"Authorization": f"Bearer {mock_context.get_auth_token_or_empty()}",
"Content-Type": "application/json",
},
)
@pytest.mark.asyncio
async def test_post_not_found_error(mock_context, mock_httpx_client):
client = HubspotCrmClient(mock_context.get_auth_token_or_empty())
mock_response = MagicMock(spec=httpx.Response)
mock_response.status_code = 404
mock_response.json.return_value = {"message": "Not Found", "errors": [{"message": "Not Found"}]}
mock_httpx_client.post.return_value = mock_response
with pytest.raises(NotFoundError) as exc_info:
await client.post("objects/contacts", json_data={"id": "123"})
assert exc_info.value.message == "Not Found"
assert exc_info.value.developer_message == '[{"message": "Not Found"}]'
@pytest.mark.asyncio
async def test_post_internal_server_error(mock_context, mock_httpx_client):
client = HubspotCrmClient(mock_context.get_auth_token_or_empty())
mock_response = MagicMock(spec=httpx.Response)
mock_response.status_code = 500
mock_response.json.return_value = {
"message": "Internal Server Error",
"errors": [{"message": "Internal Server Error"}],
}
mock_httpx_client.post.return_value = mock_response
with pytest.raises(HubspotToolExecutionError) as exc_info:
await client.post("objects/contacts", json_data={"id": "123"})
assert exc_info.value.message == "Internal Server Error"
assert exc_info.value.developer_message == '[{"message": "Internal Server Error"}]'
@pytest.mark.asyncio
@patch("arcade_hubspot.models.clean_data")
async def test_batch_get_objects(mock_clean_data, mock_context, mock_httpx_client):
mock_results = [MagicMock(spec=dict) for _ in range(3)]
clean_data_response = [MagicMock(spec=dict) for _ in range(3)]
mock_clean_data.side_effect = clean_data_response
mock_response = MagicMock(spec=httpx.Response)
mock_response.status_code = 200
mock_response.json.return_value = {
"results": mock_results,
}
mock_httpx_client.post.return_value = mock_response
client = HubspotCrmClient(mock_context.get_auth_token_or_empty())
response = await client.batch_get_objects(
HubspotObject.CONTACT, ["123", "456"], ["email", "firstname"]
)
assert response == clean_data_response
mock_clean_data.assert_has_calls([
call(result, HubspotObject.CONTACT) for result in mock_results
])
mock_httpx_client.post.assert_called_once_with(
url="https://api.hubapi.com/crm/v3/objects/contacts/batch/read",
json={
"inputs": [{"id": "123"}, {"id": "456"}],
"properties": ["email", "firstname"],
},
headers={
"Authorization": f"Bearer {mock_context.get_auth_token_or_empty()}",
"Content-Type": "application/json",
},
)
@pytest.mark.asyncio
@patch("arcade_hubspot.models.clean_data")
async def test_get_associated_objects(mock_clean_data, mock_context, mock_httpx_client):
# Mock stuff from the batch_get_objects function
mock_batch_results = [MagicMock(spec=dict) for _ in range(3)]
clean_data_response = [MagicMock(spec=dict) for _ in range(3)]
mock_clean_data.side_effect = clean_data_response
mock_post_response = MagicMock(spec=httpx.Response)
mock_post_response.status_code = 200
mock_post_response.json.return_value = {
"results": mock_batch_results,
}
mock_httpx_client.post.return_value = mock_post_response
# Mock stuff from the get_associated_objects function
mock_response = MagicMock(spec=httpx.Response)
mock_response.status_code = 200
mock_response.json.return_value = {
"results": [{"toObjectId": "123"}, {"toObjectId": "456"}],
}
mock_httpx_client.get.return_value = mock_response
client = HubspotCrmClient(mock_context.get_auth_token_or_empty())
response = await client.get_associated_objects(
parent_object=HubspotObject.COMPANY,
parent_id="parent_id",
associated_object=HubspotObject.CONTACT,
limit=5,
after=5,
properties=["email", "firstname"],
)
assert response == clean_data_response
mock_clean_data.assert_has_calls([
call(result, HubspotObject.CONTACT) for result in mock_batch_results
])
mock_httpx_client.get.assert_called_once_with(
url="https://api.hubapi.com/crm/v4/objects/company/parent_id/associations/contact",
params={
"limit": 5,
"after": 5,
},
headers={
"Authorization": f"Bearer {mock_context.get_auth_token_or_empty()}",
},
)
mock_httpx_client.post.assert_called_once_with(
url="https://api.hubapi.com/crm/v3/objects/contacts/batch/read",
json={
"inputs": [{"id": "123"}, {"id": "456"}],
"properties": ["email", "firstname"],
},
headers={
"Authorization": f"Bearer {mock_context.get_auth_token_or_empty()}",
"Content-Type": "application/json",
},
)
@pytest.mark.asyncio
@patch("arcade_hubspot.models.prepare_api_search_response")
@patch("arcade_hubspot.models.get_object_properties")
async def test_search_by_keywords(
mock_get_object_properties,
mock_prepare_api_search_response,
mock_context,
mock_httpx_client,
):
mock_response = MagicMock(spec=httpx.Response)
mock_response.status_code = 200
mock_httpx_client.post.return_value = mock_response
mock_prepared_api_response = {
"contacts": [
{
"id": "123",
"first_name": "Jason",
"last_name": "Bourne",
"email": "jason.bourne@acme.com",
}
]
}
mock_prepare_api_search_response.return_value = mock_prepared_api_response
mock_get_object_properties.return_value = ["email", "firstname"]
mock_calls = [{"id": f"call_{i}"} for i in range(3)]
mock_emails = [{"id": f"email_{i}"} for i in range(3)]
mock_notes = [{"id": f"note_{i}"} for i in range(3)]
mock_get_associated_objects = AsyncMock()
mock_get_associated_objects.side_effect = [mock_calls, mock_emails, mock_notes]
client = HubspotCrmClient(mock_context.get_auth_token_or_empty())
client.get_associated_objects = mock_get_associated_objects
response = await client.search_by_keywords(
object_type=HubspotObject.CONTACT,
keywords="test",
limit=10,
associations=[
HubspotObject.CALL,
HubspotObject.EMAIL,
HubspotObject.NOTE,
],
)
expected_response = deepcopy(mock_prepared_api_response)
expected_response["contacts"][0]["calls"] = mock_calls
expected_response["contacts"][0]["emails"] = mock_emails
expected_response["contacts"][0]["notes"] = mock_notes
assert response == expected_response
mock_httpx_client.post.assert_called_once_with(
url="https://api.hubapi.com/crm/v3/objects/contacts/search",
headers={
"Authorization": f"Bearer {mock_context.get_auth_token_or_empty()}",
"Content-Type": "application/json",
},
json={
"query": "test",
"limit": 10,
"sorts": [{"propertyName": "hs_lastmodifieddate", "direction": "DESCENDING"}],
"properties": ["email", "firstname"],
},
)
mock_get_associated_objects.assert_has_calls([
call(
parent_object=HubspotObject.CONTACT,
parent_id="123",
associated_object=HubspotObject.CALL,
limit=10,
),
call(
parent_object=HubspotObject.CONTACT,
parent_id="123",
associated_object=HubspotObject.EMAIL,
limit=10,
),
call(
parent_object=HubspotObject.CONTACT,
parent_id="123",
associated_object=HubspotObject.NOTE,
limit=10,
),
])