How to Integrate AI Agents with Pipedrive

pip install langchain langchain-openai requests python-dotenv

export PIPEDRIVE_API_TOKEN="your-pipedrive-api-token"
export PIPEDRIVE_COMPANY_DOMAIN="your-company"  # from your Pipedrive URL

import os
import requests
from datetime import datetime, timedelta
from langchain.tools import tool
from dotenv import load_dotenv

load_dotenv()

API_TOKEN = os.getenv("PIPEDRIVE_API_TOKEN")
DOMAIN = os.getenv("PIPEDRIVE_COMPANY_DOMAIN")
BASE_URL = f"https://{DOMAIN}.pipedrive.com/v1"


def pipedrive_get(endpoint: str, params: dict = None) -> dict:
    """Execute a GET request against the Pipedrive API."""
    params = params or {}
    params["api_token"] = API_TOKEN
    response = requests.get(f"{BASE_URL}/{endpoint}", params=params)
    response.raise_for_status()
    return response.json()


def pipedrive_post(endpoint: str, data: dict) -> dict:
    """Execute a POST request against the Pipedrive API."""
    response = requests.post(
        f"{BASE_URL}/{endpoint}",
        params={"api_token": API_TOKEN},
        json=data
    )
    response.raise_for_status()
    return response.json()


@tool
def search_person(name_or_email: str) -> str:
    """Search for a person in Pipedrive by name or email address."""
    result = pipedrive_get("persons/search", {
        "term": name_or_email,
        "fields": "name,email",
        "limit": 5
    })
    items = result.get("data", {}).get("items", [])
    if not items:
        return f"No person found matching '{name_or_email}'"
    formatted = []
    for item in items:
        p = item.get("item", {})
        emails = [e.get("value", "") for e in p.get("emails", [])]
        formatted.append(f"- {p.get('name', 'Unknown')} | ID: {p.get('id')} | Email: {', '.join(emails)}")
    return "Matching persons:\n" + "\n".join(formatted)


@tool
def create_deal(title: str, person_id: int = None, value: float = None,
                pipeline_stage: str = "Qualified") -> str:
    """
    Create a new deal in Pipedrive. pipeline_stage should be one of:
    'Qualified', 'Contact Made', 'Demo Scheduled', 'Proposal Made'.
    """
    # Get pipeline stages
    stages = pipedrive_get("stages")
    stage_map = {s["name"]: s["id"] for s in stages.get("data", [])}
    stage_id = stage_map.get(pipeline_stage)

    data = {"title": title}
    if person_id:
        data["person_id"] = person_id
    if value:
        data["value"] = value
    if stage_id:
        data["stage_id"] = stage_id

    result = pipedrive_post("deals", data)
    deal = result.get("data", {})
    return f"Deal created: '{deal.get('title')}' (ID: {deal.get('id')}) — Stage: {pipeline_stage}"


@tool
def add_activity(deal_id: int, subject: str, activity_type: str = "call",
                 due_date_offset_days: int = 1) -> str:
    """
    Add a follow-up activity to a deal. activity_type: call, email, meeting, task.
    due_date_offset_days: days from today until due.
    """
    due_date = (datetime.now() + timedelta(days=due_date_offset_days)).strftime("%Y-%m-%d")
    data = {
        "subject": subject,
        "type": activity_type,
        "deal_id": deal_id,
        "due_date": due_date
    }
    result = pipedrive_post("activities", data)
    activity = result.get("data", {})
    return f"Activity created: '{activity.get('subject')}' due {due_date} (ID: {activity.get('id')})"


@tool
def add_deal_note(deal_id: int, note_content: str) -> str:
    """Add a note to a Pipedrive deal with context, call summaries, or AI analysis."""
    data = {"content": note_content, "deal_id": deal_id}
    result = pipedrive_post("notes", data)
    note = result.get("data", {})
    return f"Note added to deal {deal_id} (Note ID: {note.get('id')})"


@tool
def get_stale_deals(days_inactive: int = 14) -> str:
    """List open deals with no activity in the last N days."""
    cutoff = (datetime.now() - timedelta(days=days_inactive)).strftime("%Y-%m-%d")
    deals = pipedrive_get("deals", {
        "status": "open",
        "limit": 50,
        "sort": "update_time ASC"
    })
    all_deals = deals.get("data", []) or []

    stale = []
    for deal in all_deals:
        last_activity = deal.get("last_activity_date", "")
        if not last_activity or last_activity < cutoff:
            days = (datetime.now() - datetime.strptime(
                last_activity or deal.get("add_time", "")[:10], "%Y-%m-%d"
            )).days
            owner = deal.get("owner_name", "Unassigned")
            stale.append(f"- '{deal['title']}' | Owner: {owner} | Inactive: {days} days | Value: ${deal.get('value', 0):,.0f}")

    if not stale:
        return f"No deals inactive for more than {days_inactive} days"
    return f"Stale deals (inactive {days_inactive}+ days):\n" + "\n".join(stale[:20])


@tool
def get_pipeline_summary() -> str:
    """Get a summary of the current open pipeline by stage."""
    deals = pipedrive_get("deals", {"status": "open", "limit": 500})
    all_deals = deals.get("data", []) or []

    stage_totals: dict = {}
    for deal in all_deals:
        stage = deal.get("stage_name", "Unknown")
        value = deal.get("weighted_value", 0) or 0
        if stage not in stage_totals:
            stage_totals[stage] = {"count": 0, "value": 0}
        stage_totals[stage]["count"] += 1
        stage_totals[stage]["value"] += value

    lines = ["Pipeline Summary (open deals):"]
    total_value = sum(v["value"] for v in stage_totals.values())
    for stage, data in sorted(stage_totals.items()):
        lines.append(f"  {stage}: {data['count']} deals | ${data['value']:,.0f}")
    lines.append(f"Total weighted pipeline: ${total_value:,.0f}")
    return "\n".join(lines)

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 = [search_person, create_deal, add_activity, add_deal_note,
         get_stale_deals, get_pipeline_summary]

prompt = ChatPromptTemplate.from_messages([
    ("system", """You are a sales operations specialist with access to Pipedrive CRM.

Your responsibilities:
- Qualify and enrich new leads, creating structured deals with appropriate stages
- Identify stale deals and draft follow-up activity recommendations
- Generate pipeline health summaries for sales leadership
- Add contextual notes to deals after calls or emails

When creating deals:
- Search for existing person records before creating new ones
- Set realistic stage based on lead qualification level
- Always create an immediate follow-up activity (call within 24 hours for hot leads)
- Add a qualification note summarizing what you know about the lead

When analyzing pipeline:
- Highlight deals at risk (stale, low activity, approaching close date)
- Surface quick wins (deals in late stages with high deal values)
- Flag patterns in lost deals for coaching opportunities"""),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

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

# Example: Process a new inbound lead
result = executor.invoke({
    "input": """New inbound lead received:
Name: Sarah Chen
Email: sarah.chen@techcorp.com
Company: TechCorp (500 employees, SaaS)
Message: 'We're evaluating CRM solutions for our 40-person sales team. We need pipeline visibility and forecasting. Budget is $50k/year.'

Create a qualified deal in Pipedrive, set appropriate stage, add qualification notes, and create a follow-up call activity for tomorrow."""
})
print(result["output"])

def generate_morning_pipeline_briefing() -> str:
    """Generate a daily sales pipeline briefing for the morning standup."""
    result = executor.invoke({
        "input": """Generate a concise morning pipeline briefing:
1. Get the pipeline summary by stage with deal counts and values
2. List all deals stale for 14+ days with owner names
3. Provide 3-4 bullet recommendations for the team's focus today

Format as a brief Slack message the sales team can scan in 2 minutes."""
    })
    return result["output"]

# Run daily at 8am via cron or scheduled task
briefing = generate_morning_pipeline_briefing()
print(briefing)