Building a Powerful HubSpot AI Agent: A Comprehensive Guide

Ruchir TidkeRuchir Tidke
9 min read

Introduction

Managing the core objects of HubSpot CRM - Contacts, Companies, Tickets, and Deals - is crucial for business operations. However, manually tracking and updating these objects can be a difficult task, prone to errors and inefficiencies. The static workflows and manual processes can limit your team's productivity.

This is where the HubSpot AI Agent excels. But, building a custom HubSpot agent that can intelligently interact with your CRM data and perform tasks autonomously is a challenging task, requiring extensive coding and technical expertise.

Based on our work developing custom HubSpot AI Agents for clients, we’ll guide you through the process of creating your own HubSpot agent using powerful tools integrated within the LangGraph framework. This hubspot developers tutorial offers a practical, step-by-step approach to building a fully functional AI agent capable of performing complex tasks.

Our work has shown that the HubSpot Agent, when integrated with LLMs, can intelligently interact with CRM data. It processes natural language queries, identifies relevant CRM objects, and autonomously performs operations like creating deals, updating contacts, resolving tickets, tracking deals, or generating insights. This ability to interpret user intent, call APIs, and combine multiple tools makes it much more than a CRM assistant—it’s a game-changer.

How the Hubspot Agent Works

Let’s understand the high-level overview of the workflow. This LangGraph Agent utilizes a structured workflow and Tool-calling capabilities.

  1. Query Understanding: Our Main Agent uses LLM (Large Language Model) to analyze user queries and identify the intent.

  2. Tool Calling: The agent intelligently determines the appropriate workflow and APIs or set of tools required to execute the task.

  3. Dynamic Execution: It seamlessly performs operations like creating a contact, updating a deal, or fetching CRM data.

If you are not aware of LangGraph Agents, Nodes, Tools, and Edges, then we highly recommend reading our beginner-friendly LangGraph Tutorial

Setting Up Your LangGraph Agent for HubSpot

Before diving into the main coding part, we need to follow some key steps to set up our HubSpot Dashboard. If you have already configured it, then feel free to skip this section.

  • Creating a Private App in HubSpot: Create a Private app to generate a Private Key.

  • Selecting Scopes: Define permissions to control access to CRM objects like Contacts or Deals.

  • Generating Private Key: Generate an API key to securely connect your agent with HubSpot.

  • Installing the Python HubSpot API SDK: Install and configure the hubspot-api-client Python library for interactions with HubSpot’s CRM objects.

Setting up the Environment

To get started, you need to install the below dependencies

!pip install -U hubspot-api-client langchain langgraph langchain-openai pydantic python-dotenv

Setting up API Keys

Before diving into building your Hubspot AI agent, it's crucial to set up your API keys. These keys allow your agent to interact securely with external environments and tools like Hubspot and OpenAI GPT models. Without them, the tools cannot function effectively.

Here we are using the OpenAI model, but you can use any LLM of your choice.

import getpass
import os

def _set_env(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"{var}: ")

_set_env("HUBSPOT_PRIVATE_KEY")
_set_env("OPENAI_API_KEY")

Creating the LLM Object

Here’s how to initialize the LLM using LangChain ChatOpenAI

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model_name="gpt-4o")

Creating Tools for AI Agent

We will create a few custom tools that our langgraph agent can use to interact with Hubspot CRM.

Dynamic Search Tool Integration

The Dynamic Search tool enables to fetch relevant records from CRM.

class SearchRequestSchema(BaseModel):
    object_type: str
    filterGroups: List[FilterGroup]
    properties: Optional[List[str]] = None
    limit: Optional[int] = 10
    sorts: Optional[List[str]] = None
    after: Optional[str] = None
    propertiesWithHistory: Optional[List[str]] = None
    archived: Optional[bool] = False

@tool(args_schema=SearchRequestSchema)
def dynamic_search(
    object_type: str,
    filterGroups: List[FilterGroup],
    properties: Optional[List[str]] = None,
    limit: Optional[int] = 10
):
    """Generic tool for searching objects on HubSpot Ex: contacts,deals,tickets,companies."""
    object_to_search_method = {
        'contacts': 'crm.contacts.search_api.do_search',
        'deals': 'crm.deals.search_api.do_search',
        'tickets': 'crm.tickets.search_api.do_search',
        'companies': 'crm.companies.search_api.do_search',
        'line_items': 'crm.line_items.search_api.do_search',
        'quotes': 'crm.quotes.search_api.do_search'
    }

    if object_type not in object_to_search_method:
        return {"Error": f"Unsupported object type: {object_type}"}

    try:
        api_client = get_client()
        search_method_path = object_to_search_method[object_type]
        search_method = getattr(
            api_client.crm, search_method_path.split('.')[1]
        ).search_api.do_search

        # Prepare filter groups for the request
        formatted_filter_groups = [
            {"filters": [filter.dict() for filter in group.filters]} for group in filterGroups
        ]

        # Construct the search request
        public_object_search_request = PublicObjectSearchRequest(
            filter_groups=formatted_filter_groups,
            properties=properties or [],
            limit=limit
        )

        # Send the request
        api_response = search_method(public_object_search_request=public_object_search_request)
        return {"status": "success", "response_details": api_response}
    except ApiException as e:
        return {"Exception when retrieving object information": str(e)}
    except ValueError as e:
        return {"Error": str(e)}

Output:

Create Object Tool Integration

The Create Object tool will create different crm objects and helps in associations between them as per user's questions.

class CreateObjectSchema(BaseModel):
    object_type: str
    properties: Dict[str, str]

@tool(args_schema=CreateObjectSchema)
def create_object(object_type: str, properties: Dict[str, str]):
    """ Tool for creating a CRM object (e.g., contact, deal, ticket) on HubSpot. """
    try:
        from hubspot.crm.objects import SimplePublicObjectInputForCreate

        # Mapping of object types to their respective API creation methods
        object_to_create_method = {
            'contacts': 'crm.contacts.basic_api.create',
            'deals': 'crm.deals.basic_api.create',
            'tickets': 'crm.tickets.basic_api.create',
            'companies': 'crm.companies.basic_api.create',
            'line_items': 'crm.line_items.basic_api.create',
            'quotes': 'crm.quotes.basic_api.create',
        }

        # Check if the object type is supported
        if object_type not in object_to_create_method:
            raise ValueError(f"Unsupported object type: {object_type}")

        # Initialize the API client
        api_client = get_client()

        # Get the appropriate API creation method
        create_method_path = object_to_create_method[object_type]
        create_method = getattr(api_client.crm, create_method_path.split('.')[1]).basic_api.create

        # Prepare the object input for creation
        simple_public_object_input_for_create = SimplePublicObjectInputForCreate(properties=properties)

        # Call the create method
        api_response = create_method(simple_public_object_input_for_create=simple_public_object_input_for_create)

        # Return the API response
        return {
            "status": "success",
            "response_details": api_response,
            "object_id": api_response.id
        }

    except Exception as e:
        # Handle errors and return the error message
        return {"error": str(e)}

Output:

Update Object Tool Integration

The Update Object tool will be used to update different crm objects.

class UpdateObjectSchema(BaseModel):
    object_type: str
    object_id: str
    properties: Dict[str, str]

@tool(args_schema=UpdateObjectSchema)
def update_object(object_type: str, object_id: str, properties: Dict[str, str]):
    """ Tool for updating a CRM object (e.g., contact, deal, ticket) on HubSpot using its ID. """
    try:
        from hubspot.crm.objects import SimplePublicObjectInput

        # Mapping of object types to their respective API update methods
        object_to_update_method = {
            'contacts': 'crm.contacts.basic_api.update',
            'deals': 'crm.deals.basic_api.update',
            'tickets': 'crm.tickets.basic_api.update',
            'companies': 'crm.companies.basic_api.update',
            'line_items': 'crm.line_items.basic_api.update',
            'quotes': 'crm.quotes.basic_api.update',
        }

        # Check if the object type is supported
        if object_type not in object_to_update_method:
            raise ValueError(f"Unsupported object type: {object_type}")

        # Initialize the API client
        api_client = get_client()

        # Get the appropriate API update method
        update_method_path = object_to_update_method[object_type]
        update_method = getattr(api_client.crm, update_method_path.split('.')[1]).basic_api.update

        # Prepare the object input for updating
        simple_public_object_input = SimplePublicObjectInput(properties=properties)

        # Call the update method
        api_response = update_method(object_id, simple_public_object_input)

        # Return the API response
        return {
            "status": "success",
            "response_details": api_response,
            "object_id": api_response.id
        }

    except Exception as e:
        # Handle errors and return the error message
        return {"error": str(e)}

Output:

Combining the Tools

To create a fully functional Hubspot AI agent, you need to integrate the Dynamic Search, Create Object, and Update Object tools into the LLM.

Here’s how to combine the tools:

tools = [dynamic_search, create_object, update_object]
llm_with_tools = llm.bind_tools(tools)

With this configuration, your LLM can invoke the appropriate tool based on the user’s query.

Building the LangGraph

LangGraph enables you to define a stateful workflow for your AI agent. By structuring nodes and edges, you can define how the agent processes user inputs and transitions between tools.

Steps to Build the LangGraph

  1. Define the State: Create a State dictionary to manage the agent’s inputs and outputs.

     from typing import Annotated
     from langgraph.graph import StateGraph
     from langgraph.graph.message import add_messages
     from langgraph.checkpoint.memory import MemorySaver
    
     memory = MemorySaver()
    
     class State(TypedDict):
         messages: Annotated[list, add_messages]
    
  2. Add Nodes: Add nodes for the chatbot and tools to handle user queries and invoke the tools.

     def chatbot(state: State):
          return {"messages": [llm_with_tools.invoke(state["messages"])]}
    
      graph_builder = StateGraph(State)
      graph_builder.add_node("chatbot", chatbot)
    
      tool_node = ToolNode(tools=[dynamic_search, create_object, update_object])
      graph_builder.add_node("tools", tool_node)
    
  3. Define Edges: Use conditional edges to determine when the agent should switch between nodes.

      from langgraph.prebuilt import tools_condition
    
      graph_builder.add_conditional_edges("chatbot", tools_condition)
      graph_builder.add_edge("tools", "chatbot")
      graph_builder.set_entry_point("chatbot")
    
  4. Compile the Graph: Finalize the graph for execution.

     graph = graph_builder.compile(checkpointer=memory)
    

Testing the AI Agent

Once the LangGraph is set up, you can test the agent by simulating user inputs. This ensures the tools and workflows are functioning as expected.

Interactive Testing

Run the following code to test your AI agent interactively:

config = {"configurable": {"thread_id": "1"}}

while True:
    user_input = input("User: ")
    if user_input.lower() in ["quit", "exit", "q"]:
        print("Goodbye!")
        break

    for event in graph.stream({"messages": [("user", user_input)]}, config):
        for value in event.values():
            print("Assistant:", value["messages"][-1].content)

You can provide queries like:

  • You can ask questions based on your Hubspot records.

  • "What is the current status of ‘ABC’ Deal?"

  • "Tell me the name of any two contacts with association with company ‘XYZ’. " (Trigger Search tool)

The AI agent will invoke the appropriate tool to generate responses.

Visualizing the Graph

We can easily plot a flowchart of our LangGraph Workflow.

from IPython.display import Image, display

try:
    display(Image(graph.get_graph().draw_mermaid_png()))
except Exception:
    print("Error generating graph visualization.")

Steps in the Workflow:

  1. Receive Query: A user inputs a query, such as "Update the status of Deal X to Closed Won."

  2. Interpretation: The agent uses LLMs to determine the query's intent and required action.

  3. Tool Selection: Based on the query, the agent selects the appropriate custom HubSpot AI tools.

  4. Execution: The tool is being executed and you can expect real-time updates in the Hubspot CRM.

  5. Feedback: Results are shared with the user in an easily understandable format

Testing and Refinement

Leveraging the insights we've gained from hubspot api integration services for our clients, we understand that thorough testing and refinement are essential:

  • Simulate Real-World Scenarios: Use varied queries to validate the agent’s ability to handle different tasks.

  • Validate API Operations: Ensure operations are performed correctly and efficiently.

  • Iterate and Improve: Gather user feedback and refine the agent’s workflows and capabilities over time.

While building and improving this, you may face challenges like we faced. For example, how to handle complex queries or ambiguous questions. Robust error handling can help while building this type of AI application.

Conclusion

Drawing on the experiences we’ve gained delivering tailored AI solutions to businesses, we can confidently say that the Hubspot Agent is more than just a tool—it’s an intelligent assistant that simplifies CRM management, boosts productivity, and saves valuable time. Automating repetitive tasks and enabling intelligent decision-making allows businesses to focus on what matters most—building strong customer relationships and driving growth. Looking ahead, the principles and technologies used in building the HubSpot AI Agent can also be extended to other areas such as Marketing and CMS, enabling even broader automation and intelligent data interaction across your business operations.

If you’re ready to take your HubSpot experience to the next level, building a Hubspot Agent is the way forward. It’s not just a CRM upgrade—it’s a smarter, faster way to achieve your business goals.

If you found this guide useful and want to explore more advanced techniques, don’t forget to check out our other tutorials. At FutureSmart AI, we help businesses develop state-of-the-art AI solutions tailored to their needs. To see a demo or discuss your requirements, contact us at contact@futuresmart.ai.

For real-world examples of how we’ve helped businesses, take a look at our case studies, where we showcase the practical value of our expertise.

3
Subscribe to my newsletter

Read articles from Ruchir Tidke directly inside your inbox. Subscribe to the newsletter, and don't miss out.

Written by

Ruchir Tidke
Ruchir Tidke