CRM & Customer Data Tools

Customer service becomes magical when your AI agent knows the customer. Their order history, past support tickets, preferences, and loyalty status all contribute to personalized interactions that build trust.

CRM Tool Architecture

┌──────────────────────────────────────────────────────────────────┐
│                     AI AGENT CONVERSATION                        │
│  "I want to return my order from last week"                     │
└─────────────────────────────────┬────────────────────────────────┘
                                  │
                                  ▼
┌──────────────────────────────────────────────────────────────────┐
│                     CRM MCP TOOLS                                │
├────────────────┬───────────────┬───────────────┬────────────────┤
│ get_customer   │ get_orders    │ log_interaction│ update_profile│
│ search_customers│ get_tickets  │ send_email    │ add_note       │
└────────────────┴───────────────┴───────────────┴────────────────┘
                                  │
                                  ▼
┌──────────────────────────────────────────────────────────────────┐
│                     CRM PLATFORMS                                │
│  ┌─────────┐  ┌─────────┐  ┌─────────┐  ┌─────────┐            │
│  │Salesforce│ │ HubSpot │  │  Zendesk │ │ Custom  │            │
│  └─────────┘  └─────────┘  └─────────┘  └─────────┘            │
└──────────────────────────────────────────────────────────────────┘

Core CRM Tools

1. Get Customer Profile

Retrieve comprehensive customer information:

from mcp.server import Server
from mcp import types
import json
from datetime import datetime

app = Server("crm-tools")

@app.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="get_customer",
            description="""Retrieves complete customer profile including contact info, preferences, and account status.
            Use when: Customer contacts support, places an order, or you need to personalize interaction.
            Returns: Customer profile with contact details, preferences, loyalty tier, and lifetime value.
            Note: Automatically logs access for privacy compliance.""",
            inputSchema={
                "type": "object",
                "properties": {
                    "customer_id": {
                        "type": "string",
                        "description": "Customer ID (e.g., 'cus_abc123')"
                    },
                    "email": {
                        "type": "string",
                        "format": "email",
                        "description": "Customer email address"
                    },
                    "phone": {
                        "type": "string",
                        "description": "Customer phone number"
                    },
                    "include_orders": {
                        "type": "boolean",
                        "default": True,
                        "description": "Include recent order summary"
                    },
                    "include_tickets": {
                        "type": "boolean",
                        "default": True,
                        "description": "Include open support tickets"
                    }
                },
                "anyOf": [
                    {"required": ["customer_id"]},
                    {"required": ["email"]},
                    {"required": ["phone"]}
                ]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    if name == "get_customer":
        # Find customer by any identifier
        if "customer_id" in arguments:
            customer = await crm.get_by_id(arguments["customer_id"])
        elif "email" in arguments:
            customer = await crm.get_by_email(arguments["email"])
        elif "phone" in arguments:
            customer = await crm.get_by_phone(arguments["phone"])
        else:
            return [types.TextContent(
                type="text",
                text=json.dumps({"error": "Must provide customer_id, email, or phone"})
            )]

        if not customer:
            return [types.TextContent(
                type="text",
                text=json.dumps({"found": False, "message": "Customer not found"})
            )]

        # Log access for GDPR compliance
        await audit_log.record(
            action="customer_data_access",
            customer_id=customer["id"],
            accessed_by="ai_agent",
            timestamp=datetime.utcnow()
        )

        # Build response
        profile = {
            "found": True,
            "customer_id": customer["id"],
            "name": f"{customer['first_name']} {customer['last_name']}",
            "email": customer["email"],
            "phone": customer.get("phone"),
            "created_at": customer["created_at"],
            "loyalty_tier": customer.get("loyalty_tier", "standard"),
            "lifetime_value": customer.get("ltv", 0),
            "total_orders": customer.get("order_count", 0),
            "preferences": {
                "marketing_opted_in": customer.get("marketing_consent", False),
                "preferred_contact": customer.get("contact_preference", "email"),
                "language": customer.get("language", "en")
            },
            "notes": customer.get("internal_notes", [])[-3:]  # Last 3 notes
        }

        # Include recent orders if requested
        if arguments.get("include_orders", True):
            orders = await crm.get_recent_orders(customer["id"], limit=5)
            profile["recent_orders"] = [{
                "order_id": o["id"],
                "date": o["created_at"],
                "total": o["total"],
                "status": o["status"],
                "item_count": len(o["items"])
            } for o in orders]

        # Include open tickets if requested
        if arguments.get("include_tickets", True):
            tickets = await crm.get_open_tickets(customer["id"])
            profile["open_tickets"] = [{
                "ticket_id": t["id"],
                "subject": t["subject"],
                "status": t["status"],
                "created_at": t["created_at"],
                "last_update": t["updated_at"]
            } for t in tickets]

        return [types.TextContent(type="text", text=json.dumps(profile))]

2. Get Order History

Detailed order history with filtering:

types.Tool(
    name="get_customer_orders",
    description="""Retrieves order history for a customer with filtering options.
    Use when: Customer asks about past orders, tracking, or needs help with a specific order.
    Returns: List of orders with items, status, and tracking info.""",
    inputSchema={
        "type": "object",
        "properties": {
            "customer_id": {
                "type": "string",
                "description": "Customer ID"
            },
            "status": {
                "type": "string",
                "enum": ["all", "pending", "processing", "shipped", "delivered", "cancelled", "returned"],
                "default": "all",
                "description": "Filter by order status"
            },
            "date_from": {
                "type": "string",
                "format": "date",
                "description": "Start date (YYYY-MM-DD)"
            },
            "date_to": {
                "type": "string",
                "format": "date",
                "description": "End date (YYYY-MM-DD)"
            },
            "limit": {
                "type": "integer",
                "default": 10,
                "maximum": 50,
                "description": "Maximum orders to return"
            },
            "include_items": {
                "type": "boolean",
                "default": True,
                "description": "Include line items in response"
            }
        },
        "required": ["customer_id"]
    }
)

elif name == "get_customer_orders":
    customer_id = arguments["customer_id"]
    status = arguments.get("status", "all")
    limit = arguments.get("limit", 10)
    include_items = arguments.get("include_items", True)

    # Build query filters
    filters = {"customer_id": customer_id}
    if status != "all":
        filters["status"] = status
    if "date_from" in arguments:
        filters["date_from"] = arguments["date_from"]
    if "date_to" in arguments:
        filters["date_to"] = arguments["date_to"]

    orders = await orders_db.query(filters, limit=limit)

    result = []
    for order in orders:
        order_data = {
            "order_id": order["id"],
            "order_number": order["order_number"],
            "created_at": order["created_at"],
            "status": order["status"],
            "subtotal": order["subtotal"],
            "shipping": order["shipping_cost"],
            "tax": order["tax"],
            "total": order["total"],
            "shipping_address": {
                "city": order["shipping_city"],
                "state": order["shipping_state"],
                "zip": order["shipping_zip"]
            },
            "tracking_number": order.get("tracking_number"),
            "carrier": order.get("carrier")
        }

        if include_items:
            order_data["items"] = [{
                "product_id": item["product_id"],
                "name": item["product_name"],
                "quantity": item["quantity"],
                "price": item["price"],
                "image_url": item.get("image_url")
            } for item in order["items"]]

        result.append(order_data)

    return [types.TextContent(
        type="text",
        text=json.dumps({
            "customer_id": customer_id,
            "total_found": len(result),
            "orders": result
        })
    )]

3. Log Interaction

Record customer interactions for context:

types.Tool(
    name="log_interaction",
    description="""Logs a customer interaction for future reference.
    Use when: Conversation ends, issue resolved, or important notes need recording.
    Returns: Confirmation with interaction ID.
    Note: All AI conversations should be logged for continuity.""",
    inputSchema={
        "type": "object",
        "properties": {
            "customer_id": {
                "type": "string",
                "description": "Customer ID"
            },
            "interaction_type": {
                "type": "string",
                "enum": ["chat", "email", "phone", "social", "in_app"],
                "default": "chat",
                "description": "Channel of interaction"
            },
            "summary": {
                "type": "string",
                "maxLength": 500,
                "description": "Brief summary of the interaction"
            },
            "outcome": {
                "type": "string",
                "enum": ["resolved", "escalated", "pending", "informational"],
                "description": "How the interaction ended"
            },
            "topics": {
                "type": "array",
                "items": {"type": "string"},
                "description": "Topics discussed (e.g., ['order_status', 'return'])"
            },
            "sentiment": {
                "type": "string",
                "enum": ["positive", "neutral", "negative"],
                "description": "Customer sentiment during interaction"
            },
            "related_order_id": {
                "type": "string",
                "description": "Order ID if interaction was about a specific order"
            },
            "related_ticket_id": {
                "type": "string",
                "description": "Support ticket ID if applicable"
            }
        },
        "required": ["customer_id", "summary", "outcome"]
    }
)

4. Send Customer Email

Send transactional or service emails:

types.Tool(
    name="send_customer_email",
    description="""Sends an email to a customer using predefined templates.
    Use when: Confirming actions, following up, or proactively communicating.
    Returns: Email send confirmation.
    Note: Only use approved templates - no custom HTML.""",
    inputSchema={
        "type": "object",
        "properties": {
            "customer_id": {
                "type": "string",
                "description": "Customer ID to send email to"
            },
            "template": {
                "type": "string",
                "enum": [
                    "order_confirmation",
                    "shipping_notification",
                    "delivery_confirmation",
                    "refund_processed",
                    "return_instructions",
                    "support_followup",
                    "account_update"
                ],
                "description": "Email template to use"
            },
            "template_data": {
                "type": "object",
                "description": "Dynamic data for the template (order_id, tracking_number, etc.)",
                "additionalProperties": {"type": "string"}
            },
            "reply_to_ticket": {
                "type": "string",
                "description": "Ticket ID to thread this email with"
            }
        },
        "required": ["customer_id", "template"]
    }
)

5. Update Customer Profile

Modify customer preferences or information:

types.Tool(
    name="update_customer_profile",
    description="""Updates customer profile information or preferences.
    Use when: Customer requests changes, corrects info, or updates preferences.
    Returns: Updated profile confirmation.
    Note: Logs all changes for audit trail.""",
    inputSchema={
        "type": "object",
        "properties": {
            "customer_id": {
                "type": "string",
                "description": "Customer ID to update"
            },
            "updates": {
                "type": "object",
                "properties": {
                    "first_name": {"type": "string"},
                    "last_name": {"type": "string"},
                    "phone": {"type": "string"},
                    "marketing_consent": {"type": "boolean"},
                    "preferred_contact": {
                        "type": "string",
                        "enum": ["email", "phone", "sms"]
                    },
                    "language": {"type": "string"}
                },
                "description": "Fields to update"
            },
            "reason": {
                "type": "string",
                "description": "Reason for update (for audit log)"
            }
        },
        "required": ["customer_id", "updates"]
    }
)

TypeScript Implementation

import { Server } from "@modelcontextprotocol/sdk/server";

const server = new Server({ name: "crm-tools", version: "1.0.0" });

interface Customer {
  id: string;
  first_name: string;
  last_name: string;
  email: string;
  phone?: string;
  loyalty_tier: string;
  ltv: number;
  order_count: number;
}

interface Order {
  id: string;
  order_number: string;
  status: string;
  total: number;
  items: Array<{ product_id: string; name: string; quantity: number; price: number }>;
  tracking_number?: string;
}

const crmTools = [
  {
    name: "get_customer",
    description: "Retrieves complete customer profile with contact info and preferences",
    inputSchema: {
      type: "object" as const,
      properties: {
        customer_id: { type: "string" },
        email: { type: "string", format: "email" },
        include_orders: { type: "boolean", default: true },
        include_tickets: { type: "boolean", default: true }
      }
    }
  },
  {
    name: "get_customer_orders",
    description: "Retrieves order history for a customer with filtering options",
    inputSchema: {
      type: "object" as const,
      properties: {
        customer_id: { type: "string" },
        status: {
          type: "string",
          enum: ["all", "pending", "shipped", "delivered", "cancelled"]
        },
        limit: { type: "integer", default: 10, maximum: 50 }
      },
      required: ["customer_id"]
    }
  },
  {
    name: "log_interaction",
    description: "Logs a customer interaction for future reference",
    inputSchema: {
      type: "object" as const,
      properties: {
        customer_id: { type: "string" },
        summary: { type: "string", maxLength: 500 },
        outcome: { type: "string", enum: ["resolved", "escalated", "pending"] },
        topics: { type: "array", items: { type: "string" } },
        sentiment: { type: "string", enum: ["positive", "neutral", "negative"] }
      },
      required: ["customer_id", "summary", "outcome"]
    }
  },
  {
    name: "send_customer_email",
    description: "Sends an email to a customer using predefined templates",
    inputSchema: {
      type: "object" as const,
      properties: {
        customer_id: { type: "string" },
        template: {
          type: "string",
          enum: ["order_confirmation", "shipping_notification", "refund_processed", "support_followup"]
        },
        template_data: { type: "object", additionalProperties: { type: "string" } }
      },
      required: ["customer_id", "template"]
    }
  }
];

server.setRequestHandler("tools/list", async () => ({ tools: crmTools }));

server.setRequestHandler("tools/call", async (request) => {
  const { name, arguments: args } = request.params;

  switch (name) {
    case "get_customer": {
      let customer: Customer | null = null;

      if (args.customer_id) {
        customer = await crm.getById(args.customer_id);
      } else if (args.email) {
        customer = await crm.getByEmail(args.email);
      }

      if (!customer) {
        return {
          content: [{ type: "text", text: JSON.stringify({ found: false }) }]
        };
      }

      // Log access for compliance
      await auditLog.record({
        action: "customer_data_access",
        customerId: customer.id,
        accessedBy: "ai_agent",
        timestamp: new Date()
      });

      const profile: any = {
        found: true,
        customer_id: customer.id,
        name: `${customer.first_name} ${customer.last_name}`,
        email: customer.email,
        phone: customer.phone,
        loyalty_tier: customer.loyalty_tier,
        lifetime_value: customer.ltv,
        total_orders: customer.order_count
      };

      if (args.include_orders !== false) {
        const orders = await crm.getRecentOrders(customer.id, 5);
        profile.recent_orders = orders.map((o: Order) => ({
          order_id: o.id,
          status: o.status,
          total: o.total,
          item_count: o.items.length
        }));
      }

      return {
        content: [{ type: "text", text: JSON.stringify(profile) }]
      };
    }

    case "get_customer_orders": {
      const orders = await ordersDB.getByCustomer(
        args.customer_id,
        args.status || "all",
        args.limit || 10
      );

      return {
        content: [{
          type: "text",
          text: JSON.stringify({
            customer_id: args.customer_id,
            orders: orders.map((o: Order) => ({
              order_id: o.id,
              order_number: o.order_number,
              status: o.status,
              total: o.total,
              items: o.items,
              tracking_number: o.tracking_number
            }))
          })
        }]
      };
    }

    case "log_interaction": {
      const interactionId = await crm.logInteraction({
        customerId: args.customer_id,
        type: "chat",
        summary: args.summary,
        outcome: args.outcome,
        topics: args.topics || [],
        sentiment: args.sentiment,
        agentType: "ai"
      });

      return {
        content: [{
          type: "text",
          text: JSON.stringify({
            success: true,
            interaction_id: interactionId,
            logged_at: new Date().toISOString()
          })
        }]
      };
    }

    case "send_customer_email": {
      const customer = await crm.getById(args.customer_id);
      if (!customer) {
        return {
          content: [{
            type: "text",
            text: JSON.stringify({ success: false, error: "Customer not found" })
          }]
        };
      }

      const emailId = await emailService.send({
        to: customer.email,
        template: args.template,
        data: args.template_data || {}
      });

      return {
        content: [{
          type: "text",
          text: JSON.stringify({
            success: true,
            email_id: emailId,
            sent_to: customer.email,
            template: args.template
          })
        }]
      };
    }

    default:
      throw new Error(`Unknown tool: ${name}`);
  }
});

Privacy and Compliance

Required Compliance Features

# 1. Log all data access
await audit_log.record(
    action="customer_data_access",
    customer_id=customer_id,
    fields_accessed=["email", "phone", "orders"],
    accessed_by="ai_agent",
    purpose="support_request",
    timestamp=datetime.utcnow()
)

# 2. Respect marketing consent
if not customer["marketing_consent"]:
    # Don't send promotional emails
    raise PermissionError("Customer has not consented to marketing")

# 3. Data deletion tool (GDPR right to erasure)
types.Tool(
    name="request_data_deletion",
    description="Initiates GDPR data deletion request for a customer",
    inputSchema={
        "type": "object",
        "properties": {
            "customer_id": {"type": "string"},
            "reason": {"type": "string"}
        },
        "required": ["customer_id"]
    }
)

Real-World Conversation Flow

Customer: "Hi, I want to check on my recent order"

AI Agent: [Calls get_customer]
  get_customer(email="sarah@example.com", include_orders=True)

Tool Response:
{
  "found": true,
  "customer_id": "cus_abc123",
  "name": "Sarah Johnson",
  "loyalty_tier": "gold",
  "recent_orders": [
    {
      "order_id": "ord_xyz",
      "date": "2025-01-10",
      "total": 149.99,
      "status": "shipped",
      "tracking_number": "1Z999AA10123456784"
    }
  ]
}

AI Agent: "Hi Sarah! I can see you're a Gold member - thank you for
your loyalty! I found your recent order from January 10th for $149.99.
It's already shipped! Here's your tracking number: 1Z999AA10123456784.

Would you like me to check the current delivery status?"

Customer: "Yes please"

AI Agent: [Calls track_shipment]
  track_shipment(tracking_number="1Z999AA10123456784")

AI Agent: "Great news! Your package is out for delivery today and
should arrive by 6 PM. Is there anything else I can help with?"

Customer: "No, that's all. Thanks!"

AI Agent: [Calls log_interaction]
  log_interaction(
    customer_id="cus_abc123",
    summary="Customer inquired about order status. Package out for delivery.",
    outcome="resolved",
    topics=["order_status", "tracking"],
    sentiment="positive"
  )

AI Agent: "You're welcome, Sarah! Have a great day!"

Key Takeaways

Next up: Analytics & Reporting Tools - Building tools to query sales data, generate reports, and surface business insights.