"""HubSpot MCP Server.
Provides CRM tools for contacts, companies, and deals via HubSpot API v3.
All authentication is handled by the MCP Gateway via OAuth 2.0.
"""
import logging
import os
from typing import Dict, List, Optional
from pydantic import Field
from mcp_base import (
create_base_app,
BaseMCPSettings,
run_server,
AuthenticationError,
extract_token_from_headers,
)
from hubspot_mcp.client import HubSpotClient
LOGGER = logging.getLogger(__name__)
class HubSpotSettings(BaseMCPSettings):
"""Settings for HubSpot MCP server."""
name: str = Field(default="HubSpot")
version: str = Field(default="0.1.0")
description: str = Field(default="HubSpot CRM tools for contacts, companies, and deals")
port: int = Field(default=3004)
def create_app():
"""Create the HubSpot MCP server."""
settings = HubSpotSettings()
mcp = create_base_app(settings, register_health=True)
# Register HubSpot tools
_register_contact_tools(mcp, settings)
_register_company_tools(mcp, settings)
_register_deal_tools(mcp, settings)
_register_pipeline_tools(mcp, settings)
_register_owner_tools(mcp, settings)
_register_list_tools(mcp, settings)
_register_engagement_tools(mcp, settings)
_register_association_tools(mcp, settings)
_register_property_tools(mcp, settings)
tool_count = len(list(mcp._tool_manager._tools.keys()))
LOGGER.info(f"🟠 HubSpot initialized with {tool_count} tools")
return mcp, settings
def _get_client(headers: dict) -> HubSpotClient:
"""Get authenticated HubSpot client from headers.
Uses mcp_base.extract_token_from_headers() to extract the bearer token
from the _headers dict injected by the gateway.
Args:
headers: Dict containing authorization header from gateway
Returns:
Authenticated HubSpotClient instance
Raises:
AuthenticationError: If no valid token is found
"""
access_token = extract_token_from_headers(
headers or {},
fallback_env="HUBSPOT_ACCESS_TOKEN",
)
if not access_token:
raise AuthenticationError(
"No HubSpot access token available. "
"Please authenticate via the gateway's OAuth flow at /oauth/hubspot/authorize"
)
return HubSpotClient(access_token=access_token)
# -----------------------------------------------------------------------------
# Contact Tools
# -----------------------------------------------------------------------------
def _register_contact_tools(mcp, settings):
"""Register HubSpot contact tools."""
@mcp.tool()
def hubspot_list_contacts(
limit: int = Field(default=50, description="Maximum contacts to return (1-100)"),
properties: Optional[List[str]] = Field(default=None, description="Properties to include"),
after: Optional[str] = Field(default=None, description="Pagination cursor"),
_headers: dict = None,
) -> dict:
"""List HubSpot contacts with pagination.
Returns contacts with their properties. Use 'after' cursor for pagination.
Args:
_headers: Request headers (automatically injected by gateway)
"""
client = _get_client(_headers or {})
return client.list_contacts(limit=limit, properties=properties, after=after)
@mcp.tool()
def hubspot_get_contact(
contact_id: str = Field(..., description="Contact ID or email"),
properties: Optional[List[str]] = Field(default=None, description="Properties to include"),
_headers: dict = None,
) -> dict:
"""Get a specific HubSpot contact by ID.
Returns the contact with all requested properties.
Args:
_headers: Request headers (automatically injected by gateway)
"""
client = _get_client(_headers or {})
return client.get_contact(contact_id=contact_id, properties=properties)
@mcp.tool()
def hubspot_create_contact(
email: str = Field(..., description="Contact email address"),
firstname: Optional[str] = Field(default=None, description="First name"),
lastname: Optional[str] = Field(default=None, description="Last name"),
phone: Optional[str] = Field(default=None, description="Phone number"),
company: Optional[str] = Field(default=None, description="Company name"),
_headers: dict = None,
) -> dict:
"""Create a new HubSpot contact.
Creates a contact with the provided properties. Email is required.
Returns the created contact with its ID.
Args:
_headers: Request headers (automatically injected by gateway)
"""
properties = {"email": email}
if firstname:
properties["firstname"] = firstname
if lastname:
properties["lastname"] = lastname
if phone:
properties["phone"] = phone
if company:
properties["company"] = company
client = _get_client(_headers or {})
return client.create_contact(properties=properties)
@mcp.tool()
def hubspot_update_contact(
contact_id: str = Field(..., description="Contact ID to update"),
email: Optional[str] = Field(default=None, description="New email"),
firstname: Optional[str] = Field(default=None, description="New first name"),
lastname: Optional[str] = Field(default=None, description="New last name"),
phone: Optional[str] = Field(default=None, description="New phone number"),
_headers: dict = None,
) -> dict:
"""Update an existing HubSpot contact.
Only provided fields will be updated.
Args:
_headers: Request headers (automatically injected by gateway)
"""
properties = {}
if email is not None:
properties["email"] = email
if firstname is not None:
properties["firstname"] = firstname
if lastname is not None:
properties["lastname"] = lastname
if phone is not None:
properties["phone"] = phone
client = _get_client(_headers or {})
return client.update_contact(contact_id=contact_id, properties=properties)
@mcp.tool()
def hubspot_search_contacts(
query: str = Field(..., description="Search query (name, email, etc.)"),
limit: int = Field(default=10, description="Maximum results (1-100)"),
_headers: dict = None,
) -> dict:
"""Search HubSpot contacts.
Searches across contact properties including name, email, and company.
Args:
_headers: Request headers (automatically injected by gateway)
"""
client = _get_client(_headers or {})
return client.search_contacts(query=query, limit=limit)
LOGGER.info("✅ Contact tools registered (5 tools)")
# -----------------------------------------------------------------------------
# Company Tools
# -----------------------------------------------------------------------------
def _register_company_tools(mcp, settings):
"""Register HubSpot company tools."""
@mcp.tool()
def hubspot_list_companies(
limit: int = Field(default=50, description="Maximum companies to return (1-100)"),
properties: Optional[List[str]] = Field(default=None, description="Properties to include"),
after: Optional[str] = Field(default=None, description="Pagination cursor"),
_headers: dict = None,
) -> dict:
"""List HubSpot companies with pagination.
Args:
_headers: Request headers (automatically injected by gateway)
"""
client = _get_client(_headers or {})
return client.list_companies(limit=limit, properties=properties, after=after)
@mcp.tool()
def hubspot_get_company(
company_id: str = Field(..., description="Company ID"),
properties: Optional[List[str]] = Field(default=None, description="Properties to include"),
_headers: dict = None,
) -> dict:
"""Get a specific HubSpot company by ID.
Args:
_headers: Request headers (automatically injected by gateway)
"""
client = _get_client(_headers or {})
return client.get_company(company_id=company_id, properties=properties)
@mcp.tool()
def hubspot_create_company(
name: str = Field(..., description="Company name"),
domain: Optional[str] = Field(default=None, description="Company website domain"),
industry: Optional[str] = Field(default=None, description="Industry"),
description: Optional[str] = Field(default=None, description="Company description"),
_headers: dict = None,
) -> dict:
"""Create a new HubSpot company.
Creates a company with the provided properties. Name is required.
Args:
_headers: Request headers (automatically injected by gateway)
"""
properties = {"name": name}
if domain:
properties["domain"] = domain
if industry:
properties["industry"] = industry
if description:
properties["description"] = description
client = _get_client(_headers or {})
return client.create_company(properties=properties)
@mcp.tool()
def hubspot_update_company(
company_id: str = Field(..., description="Company ID to update"),
name: Optional[str] = Field(default=None, description="New company name"),
domain: Optional[str] = Field(default=None, description="New website domain"),
industry: Optional[str] = Field(default=None, description="New industry"),
description: Optional[str] = Field(default=None, description="New description"),
_headers: dict = None,
) -> dict:
"""Update an existing HubSpot company.
Only provided fields will be updated.
Args:
_headers: Request headers (automatically injected by gateway)
"""
properties = {}
if name is not None:
properties["name"] = name
if domain is not None:
properties["domain"] = domain
if industry is not None:
properties["industry"] = industry
if description is not None:
properties["description"] = description
client = _get_client(_headers or {})
return client.update_company(company_id=company_id, properties=properties)
@mcp.tool()
def hubspot_search_companies(
query: str = Field(..., description="Search query (name, domain, etc.)"),
limit: int = Field(default=10, description="Maximum results (1-100)"),
_headers: dict = None,
) -> dict:
"""Search HubSpot companies.
Args:
_headers: Request headers (automatically injected by gateway)
"""
client = _get_client(_headers or {})
return client.search_companies(query=query, limit=limit)
LOGGER.info("✅ Company tools registered (5 tools)")
# -----------------------------------------------------------------------------
# Deal Tools
# -----------------------------------------------------------------------------
def _register_deal_tools(mcp, settings):
"""Register HubSpot deal tools."""
@mcp.tool()
def hubspot_list_deals(
limit: int = Field(default=50, description="Maximum deals to return (1-100)"),
properties: Optional[List[str]] = Field(default=None, description="Properties to include"),
after: Optional[str] = Field(default=None, description="Pagination cursor"),
_headers: dict = None,
) -> dict:
"""List HubSpot deals with pagination.
Args:
_headers: Request headers (automatically injected by gateway)
"""
client = _get_client(_headers or {})
return client.list_deals(limit=limit, properties=properties, after=after)
@mcp.tool()
def hubspot_get_deal(
deal_id: str = Field(..., description="Deal ID"),
properties: Optional[List[str]] = Field(default=None, description="Properties to include"),
_headers: dict = None,
) -> dict:
"""Get a specific HubSpot deal by ID.
Args:
_headers: Request headers (automatically injected by gateway)
"""
client = _get_client(_headers or {})
return client.get_deal(deal_id=deal_id, properties=properties)
@mcp.tool()
def hubspot_create_deal(
dealname: str = Field(..., description="Deal name"),
amount: Optional[str] = Field(default=None, description="Deal amount"),
dealstage: Optional[str] = Field(default=None, description="Deal stage ID"),
pipeline: Optional[str] = Field(default=None, description="Pipeline ID"),
closedate: Optional[str] = Field(default=None, description="Expected close date (ISO format)"),
_headers: dict = None,
) -> dict:
"""Create a new HubSpot deal.
Creates a deal with the provided properties. Deal name is required.
Args:
_headers: Request headers (automatically injected by gateway)
"""
properties = {"dealname": dealname}
if amount:
properties["amount"] = amount
if dealstage:
properties["dealstage"] = dealstage
if pipeline:
properties["pipeline"] = pipeline
if closedate:
properties["closedate"] = closedate
client = _get_client(_headers or {})
return client.create_deal(properties=properties)
@mcp.tool()
def hubspot_update_deal(
deal_id: str = Field(..., description="Deal ID to update"),
dealname: Optional[str] = Field(default=None, description="New deal name"),
amount: Optional[str] = Field(default=None, description="New amount"),
dealstage: Optional[str] = Field(default=None, description="New deal stage ID"),
closedate: Optional[str] = Field(default=None, description="New close date"),
_headers: dict = None,
) -> dict:
"""Update an existing HubSpot deal.
Only provided fields will be updated.
Args:
_headers: Request headers (automatically injected by gateway)
"""
properties = {}
if dealname is not None:
properties["dealname"] = dealname
if amount is not None:
properties["amount"] = amount
if dealstage is not None:
properties["dealstage"] = dealstage
if closedate is not None:
properties["closedate"] = closedate
client = _get_client(_headers or {})
return client.update_deal(deal_id=deal_id, properties=properties)
@mcp.tool()
def hubspot_search_deals(
query: str = Field(..., description="Search query (deal name, amount, etc.)"),
limit: int = Field(default=10, description="Maximum results (1-100)"),
_headers: dict = None,
) -> dict:
"""Search HubSpot deals.
Searches across deal properties including name, amount, and stage.
Args:
_headers: Request headers (automatically injected by gateway)
"""
client = _get_client(_headers or {})
return client.search_deals(query=query, limit=limit)
LOGGER.info("✅ Deal tools registered (5 tools)")
# -----------------------------------------------------------------------------
# Pipeline Tools
# -----------------------------------------------------------------------------
def _register_pipeline_tools(mcp, settings):
"""Register HubSpot pipeline tools."""
@mcp.tool()
def hubspot_list_pipelines(
object_type: str = Field(default="deals", description="Object type: deals or tickets"),
_headers: dict = None,
) -> dict:
"""List HubSpot pipelines for an object type.
Returns all pipelines and their stages for the specified object type.
Args:
_headers: Request headers (automatically injected by gateway)
"""
client = _get_client(_headers or {})
return client.list_pipelines(object_type=object_type)
@mcp.tool()
def hubspot_get_pipeline(
pipeline_id: str = Field(..., description="Pipeline ID"),
object_type: str = Field(default="deals", description="Object type: deals or tickets"),
_headers: dict = None,
) -> dict:
"""Get a specific HubSpot pipeline by ID.
Returns the pipeline details including all stages.
Args:
_headers: Request headers (automatically injected by gateway)
"""
client = _get_client(_headers or {})
return client.get_pipeline(pipeline_id=pipeline_id, object_type=object_type)
LOGGER.info("✅ Pipeline tools registered (2 tools)")
# -----------------------------------------------------------------------------
# Owner Tools
# -----------------------------------------------------------------------------
def _register_owner_tools(mcp, settings):
"""Register HubSpot owner tools."""
@mcp.tool()
def hubspot_list_owners(
limit: int = Field(default=100, description="Maximum owners to return"),
_headers: dict = None,
) -> dict:
"""List HubSpot owners (users).
Returns all users/owners in the HubSpot account.
Useful for assigning contacts, companies, or deals to users.
Args:
_headers: Request headers (automatically injected by gateway)
"""
client = _get_client(_headers or {})
return client.list_owners(limit=limit)
@mcp.tool()
def hubspot_get_owner(
owner_id: str = Field(..., description="Owner ID"),
_headers: dict = None,
) -> dict:
"""Get a specific HubSpot owner by ID.
Args:
_headers: Request headers (automatically injected by gateway)
"""
client = _get_client(_headers or {})
return client.get_owner(owner_id=owner_id)
LOGGER.info("✅ Owner tools registered (2 tools)")
# -----------------------------------------------------------------------------
# Contact List Tools
# -----------------------------------------------------------------------------
def _register_list_tools(mcp, settings):
"""Register HubSpot contact list tools."""
@mcp.tool()
def hubspot_list_contact_lists(
limit: int = Field(default=100, description="Maximum lists to return"),
_headers: dict = None,
) -> dict:
"""List all HubSpot contact lists.
Returns all static and dynamic (smart) lists in the account.
Args:
_headers: Request headers (automatically injected by gateway)
"""
client = _get_client(_headers or {})
return client.list_contact_lists(limit=limit)
@mcp.tool()
def hubspot_get_contact_list(
list_id: str = Field(..., description="Contact list ID"),
_headers: dict = None,
) -> dict:
"""Get a specific HubSpot contact list by ID.
Args:
_headers: Request headers (automatically injected by gateway)
"""
client = _get_client(_headers or {})
return client.get_contact_list(list_id=list_id)
@mcp.tool()
def hubspot_create_contact_list(
name: str = Field(..., description="Name of the new list"),
dynamic: bool = Field(default=False, description="True for smart list, False for static"),
_headers: dict = None,
) -> dict:
"""Create a new HubSpot contact list.
Creates a static or dynamic (smart) contact list.
Static lists require manually adding contacts.
Dynamic lists automatically update based on filters.
Args:
_headers: Request headers (automatically injected by gateway)
"""
client = _get_client(_headers or {})
return client.create_contact_list(name=name, dynamic=dynamic)
@mcp.tool()
def hubspot_add_contacts_to_list(
list_id: str = Field(..., description="Contact list ID"),
contact_ids: List[int] = Field(..., description="List of contact IDs (integers) to add"),
_headers: dict = None,
) -> dict:
"""Add contacts to a static HubSpot list.
Only works with static lists, not dynamic (smart) lists.
Args:
_headers: Request headers (automatically injected by gateway)
"""
client = _get_client(_headers or {})
return client.add_contacts_to_list(list_id=list_id, contact_ids=contact_ids)
@mcp.tool()
def hubspot_remove_contacts_from_list(
list_id: str = Field(..., description="Contact list ID"),
contact_ids: List[int] = Field(..., description="List of contact IDs (integers) to remove"),
_headers: dict = None,
) -> dict:
"""Remove contacts from a static HubSpot list.
Only works with static lists, not dynamic (smart) lists.
Args:
_headers: Request headers (automatically injected by gateway)
"""
client = _get_client(_headers or {})
return client.remove_contacts_from_list(list_id=list_id, contact_ids=contact_ids)
LOGGER.info("✅ Contact list tools registered (5 tools)")
# -----------------------------------------------------------------------------
# Engagement Tools
# -----------------------------------------------------------------------------
def _register_engagement_tools(mcp, settings):
"""Register HubSpot engagement/activity tools."""
@mcp.tool()
def hubspot_create_note(
body: str = Field(..., description="Note content (supports HTML)"),
contact_ids: Optional[List[int]] = Field(default=None, description="Contact IDs to associate"),
company_ids: Optional[List[int]] = Field(default=None, description="Company IDs to associate"),
deal_ids: Optional[List[int]] = Field(default=None, description="Deal IDs to associate"),
_headers: dict = None,
) -> dict:
"""Create a note in HubSpot.
Creates a note and associates it with the specified records.
At least one association (contact, company, or deal) should be provided.
Args:
_headers: Request headers (automatically injected by gateway)
"""
client = _get_client(_headers or {})
return client.create_note(
body=body,
contact_ids=contact_ids,
company_ids=company_ids,
deal_ids=deal_ids
)
@mcp.tool()
def hubspot_create_task(
subject: str = Field(..., description="Task subject/title"),
body: str = Field(default="", description="Task description"),
contact_ids: Optional[List[int]] = Field(default=None, description="Contact IDs to associate"),
company_ids: Optional[List[int]] = Field(default=None, description="Company IDs to associate"),
deal_ids: Optional[List[int]] = Field(default=None, description="Deal IDs to associate"),
_headers: dict = None,
) -> dict:
"""Create a task in HubSpot.
Creates a task and associates it with the specified records.
Tasks appear in the HubSpot tasks queue for assignment and tracking.
Args:
_headers: Request headers (automatically injected by gateway)
"""
client = _get_client(_headers or {})
return client.create_task(
subject=subject,
body=body,
contact_ids=contact_ids,
company_ids=company_ids,
deal_ids=deal_ids
)
LOGGER.info("✅ Engagement tools registered (2 tools)")
# -----------------------------------------------------------------------------
# Association Tools
# -----------------------------------------------------------------------------
def _register_association_tools(mcp, settings):
"""Register HubSpot association tools."""
@mcp.tool()
def hubspot_create_association(
from_object_type: str = Field(..., description="Source object type (contacts, companies, deals)"),
from_object_id: str = Field(..., description="Source object ID"),
to_object_type: str = Field(..., description="Target object type (contacts, companies, deals)"),
to_object_id: str = Field(..., description="Target object ID"),
association_type: str = Field(default="contact_to_company", description="Association type (e.g., contact_to_company, deal_to_company)"),
_headers: dict = None,
) -> dict:
"""Create an association between two HubSpot CRM objects.
Links two objects (e.g., link a contact to a company, or a deal to a company).
Common association types:
- contact_to_company
- company_to_contact
- deal_to_company
- company_to_deal
Args:
_headers: Request headers (automatically injected by gateway)
"""
client = _get_client(_headers or {})
return client.create_association(
from_object_type=from_object_type,
from_object_id=from_object_id,
to_object_type=to_object_type,
to_object_id=to_object_id,
association_type=association_type
)
@mcp.tool()
def hubspot_get_associations(
object_type: str = Field(..., description="Object type (contacts, companies, deals)"),
object_id: str = Field(..., description="Object ID"),
to_object_type: str = Field(..., description="Target object type to get associations for"),
_headers: dict = None,
) -> dict:
"""Get associations for a HubSpot CRM object.
Returns all objects of the target type associated with the source object.
Args:
_headers: Request headers (automatically injected by gateway)
"""
client = _get_client(_headers or {})
return client.get_associations(
object_type=object_type,
object_id=object_id,
to_object_type=to_object_type
)
LOGGER.info("✅ Association tools registered (2 tools)")
# -----------------------------------------------------------------------------
# Property Tools
# -----------------------------------------------------------------------------
def _register_property_tools(mcp, settings):
"""Register HubSpot property tools."""
@mcp.tool()
def hubspot_list_properties(
object_type: str = Field(default="contacts", description="Object type: contacts, companies, or deals"),
_headers: dict = None,
) -> dict:
"""List all properties for a HubSpot object type.
Returns all custom and default properties including their types and options.
Args:
_headers: Request headers (automatically injected by gateway)
"""
client = _get_client(_headers or {})
return client.list_properties(object_type=object_type)
@mcp.tool()
def hubspot_get_property(
object_type: str = Field(..., description="Object type: contacts, companies, or deals"),
property_name: str = Field(..., description="Property internal name"),
_headers: dict = None,
) -> dict:
"""Get details of a specific HubSpot property.
Returns property definition including type, label, and options.
Args:
_headers: Request headers (automatically injected by gateway)
"""
client = _get_client(_headers or {})
return client.get_property(object_type=object_type, property_name=property_name)
LOGGER.info("✅ Property tools registered (2 tools)")
if __name__ == "__main__":
mcp, settings = create_app()
run_server(mcp, settings)
