How to Integrate AI Agents with Intercom

pip install langchain langchain-openai requests python-dotenv

export INTERCOM_ACCESS_TOKEN="your-intercom-access-token"

import os
import requests
from langchain.tools import tool
from dotenv import load_dotenv

load_dotenv()

INTERCOM_TOKEN = os.getenv("INTERCOM_ACCESS_TOKEN")
HEADERS = {
    "Authorization": f"Bearer {INTERCOM_TOKEN}",
    "Accept": "application/json",
    "Content-Type": "application/json",
    "Intercom-Version": "2.10"
}
BASE_URL = "https://api.intercom.io"


@tool
def get_conversation(conversation_id: str) -> str:
    """Get full details of an Intercom conversation including messages and customer info."""
    response = requests.get(f"{BASE_URL}/conversations/{conversation_id}", headers=HEADERS)
    conv = response.json()

    messages = conv.get("conversation_parts", {}).get("conversation_parts", [])
    recent_messages = []
    for msg in messages[-5:]:  # Last 5 messages
        author_type = msg.get("author", {}).get("type", "unknown")
        body = msg.get("body", "").replace("<p>", "").replace("</p>", "\n")
        recent_messages.append(f"[{author_type}]: {body[:300]}")

    customer = conv.get("source", {}).get("author", {})
    return f"""Conversation {conversation_id}:
Customer: {customer.get('name', 'Unknown')} ({customer.get('email', 'no email')})
State: {conv.get('state', 'unknown')}
Recent messages:
{chr(10).join(recent_messages)}"""


@tool
def add_internal_note(conversation_id: str, note_text: str) -> str:
    """Add an internal note (visible only to support agents) to an Intercom conversation."""
    data = {
        "message_type": "note",
        "type": "admin",
        "body": f"<p>{note_text}</p>"
    }
    response = requests.post(
        f"{BASE_URL}/conversations/{conversation_id}/reply",
        headers=HEADERS,
        json=data
    )
    return "Internal note added" if response.status_code == 200 else f"Error: {response.text}"


@tool
def add_conversation_tags(conversation_id: str, tag_names: list) -> str:
    """Add tags to an Intercom conversation for classification and routing."""
    # First get tag IDs
    tags_response = requests.get(f"{BASE_URL}/tags", headers=HEADERS)
    existing_tags = {t["name"]: t["id"] for t in tags_response.json().get("data", [])}

    tagged = []
    for tag_name in tag_names:
        if tag_name in existing_tags:
            requests.post(
                f"{BASE_URL}/conversations/{conversation_id}/tags",
                headers=HEADERS,
                json={"id": existing_tags[tag_name]}
            )
            tagged.append(tag_name)
    return f"Tags added: {tagged}" if tagged else "No matching tags found"


@tool
def search_help_center(query: str) -> str:
    """Search Intercom Help Center articles for content relevant to a customer query."""
    response = requests.get(
        f"{BASE_URL}/articles",
        headers=HEADERS,
        params={"q": query, "per_page": 5}
    )
    articles = response.json().get("data", [])
    if not articles:
        return f"No help center articles found for: '{query}'"

    result = []
    for article in articles[:3]:
        title = article.get("title", "Untitled")
        url = article.get("url", "")
        result.append(f"- [{title}]({url})")
    return "Relevant help center articles:\n" + "\n".join(result)


@tool
def list_open_conversations(team_name: str = None) -> str:
    """List recent open conversations, optionally filtered by team."""
    params = {"state": "open", "display_as": "plaintext", "per_page": 20}
    response = requests.get(f"{BASE_URL}/conversations", headers=HEADERS, params=params)
    convs = response.json().get("data", [])

    result = [f"Open conversations ({len(convs)} shown):"]
    for conv in convs[:10]:
        customer = conv.get("source", {}).get("author", {}).get("email", "unknown")
        waiting_since = conv.get("waiting_since", 0)
        result.append(f"- ID: {conv['id']} | Customer: {customer} | Waiting: {waiting_since}s")
    return "\n".join(result)

from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o", temperature=0.1)
tools = [get_conversation, add_internal_note, add_conversation_tags,
         search_help_center, list_open_conversations]

prompt = ChatPromptTemplate.from_messages([
    ("system", """You are a customer support specialist assistant with access to Intercom.

When a new conversation arrives:
1. Read the conversation to understand the customer's issue
2. Search the help center for relevant articles
3. Add appropriate classification tags (billing, technical, onboarding, general)
4. Draft a helpful reply and add it as an internal note for the human agent
5. If urgent (payment issue, data loss, security), flag it prominently

Draft reply guidelines:
- Be empathetic and professional
- Include specific, actionable next steps
- Link to relevant help center articles
- Keep replies concise and clear
- Never make promises you can't keep"""),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True, max_iterations=6)

# Process a new conversation
result = executor.invoke({
    "input": "Process new conversation ID: 123456789. Classify it, find relevant help articles, and draft a reply for the human agent."
})
print(result["output"])

from flask import Flask, request, jsonify
import hmac, hashlib

app = Flask(__name__)
INTERCOM_CLIENT_SECRET = os.getenv("INTERCOM_CLIENT_SECRET")


def verify_intercom_signature(payload: bytes, signature: str) -> bool:
    expected = hmac.new(
        INTERCOM_CLIENT_SECRET.encode(),
        payload,
        hashlib.sha1
    ).hexdigest()
    return hmac.compare_digest(f"sha1={expected}", signature)


@app.route('/intercom/webhook', methods=['POST'])
def intercom_webhook():
    signature = request.headers.get('X-Hub-Signature', '')
    if not verify_intercom_signature(request.data, signature):
        return jsonify({"error": "Invalid signature"}), 401

    event = request.json
    topic = event.get("topic", "")
    data = event.get("data", {})

    if topic == "conversation.user.created":
        conversation_id = data.get("item", {}).get("id")
        if conversation_id:
            # Triage in background thread
            import threading
            def triage():
                executor.invoke({
                    "input": f"Triage new conversation {conversation_id}: classify, search help center, add note with draft reply"
                })
            threading.Thread(target=triage).start()

    return jsonify({"received": True})