Model Context Protocol (MCP)

Extend your assistants with custom functions and external integrations using Plati’s Model Context Protocol.

What is MCP?: The Model Context Protocol allows your assistants to call external functions, query databases, integrate with APIs, and perform custom business logic in real-time during conversations.

Understanding MCP

MCP bridges the gap between your AI assistants and the external world:

What MCP Enables

Query external databases
Call third-party APIs
Send notifications or emails
Perform custom business logic
Access real-time data
Integrate with any system

How It Works

You create HTTP server with MCP protocol
Register functions in your Plati workspace
Connect functions to specific assistants
Functions become available in prompts

MCP Architecture Flow

Building Your First MCP Server

1. Basic HTTP Server

Create an HTTP server that implements the MCP protocol:

from flask import Flask, request, jsonify
import requests

app = Flask(__name__)

@app.route('/tools', methods=['GET'])
def list_tools():
    """Return available functions for this MCP"""
    return jsonify({
        "tools": [
            {
                "name": "check_order_status",
                "description": "Check the status of a customer order",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "order_id": {
                            "type": "string",
                            "description": "The order ID to check"
                        }
                    },
                    "required": ["order_id"]
                }
            },
            {
                "name": "send_email",
                "description": "Send an email to a customer",
                "input_schema": {
                    "type": "object", 
                    "properties": {
                        "to": {"type": "string"},
                        "subject": {"type": "string"},
                        "body": {"type": "string"}
                    },
                    "required": ["to", "subject", "body"]
                }
            }
        ]
    })

@app.route('/call', methods=['POST'])
def call_function():
    """Execute a function call"""
    data = request.get_json()
    function_name = data.get('name')
    arguments = data.get('arguments', {})
    
    if function_name == 'check_order_status':
        return check_order_status(arguments)
    elif function_name == 'send_email':
        return send_email(arguments)
    else:
        return jsonify({"error": "Function not found"}), 404

def check_order_status(args):
    order_id = args.get('order_id')
    # Mock implementation - replace with your actual logic
    orders_db = {
        "ORD-001": {"status": "shipped", "tracking": "1Z999AA1234567890"},
        "ORD-002": {"status": "processing", "estimated_ship": "2024-01-15"}
    }
    
    order = orders_db.get(order_id)
    if order:
        return jsonify({"success": True, "data": order})
    else:
        return jsonify({"success": False, "error": "Order not found"})

def send_email(args):
    # Mock implementation - integrate with your email service
    print(f"Sending email to {args['to']}: {args['subject']}")
    return jsonify({"success": True, "message": "Email sent successfully"})

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

2. Register Your MCP in Plati

Once your server is running, register it with Plati:

curl -X POST https://api.plati.ai/workspace/mcps \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Order Management System",
    "transport": "http",
    "url": "https://your-domain.com/mcp",
    "headers": [
      {
        "key": "Authorization",
        "value": "Bearer YOUR_MCP_API_KEY"
      }
    ]
  }'

3. Connect MCP to Assistant

Link your MCP functions to a specific assistant:

curl -X PUT https://api.plati.ai/assistant/ASSISTANT_ID/mcp/MCP_ID \
  -H "Authorization: Bearer YOUR_API_KEY"

Real-World MCP Examples

E-commerce Integration

from flask import Flask, request, jsonify
import sqlite3
from datetime import datetime

app = Flask(__name__)

@app.route('/tools', methods=['GET'])
def list_tools():
    return jsonify({
        "tools": [
            {
                "name": "check_order_status",
                "description": "Get current order status and tracking info",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "order_id": {"type": "string"}
                    },
                    "required": ["order_id"]
                }
            },
            {
                "name": "process_return",
                "description": "Initiate a return for an order",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "order_id": {"type": "string"},
                        "reason": {"type": "string"},
                        "items": {"type": "array"}
                    },
                    "required": ["order_id", "reason"]
                }
            },
            {
                "name": "apply_discount",
                "description": "Apply a discount code to customer account",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "customer_id": {"type": "string"},
                        "discount_code": {"type": "string"}
                    },
                    "required": ["customer_id", "discount_code"]
                }
            },
            {
                "name": "get_product_recommendations",
                "description": "Get personalized product recommendations",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "customer_id": {"type": "string"},
                        "category": {"type": "string"},
                        "limit": {"type": "integer", "default": 5}
                    },
                    "required": ["customer_id"]
                }
            }
        ]
    })

@app.route('/call', methods=['POST'])
def call_function():
    data = request.get_json()
    function_name = data.get('name')
    arguments = data.get('arguments', {})
    
    function_map = {
        'check_order_status': check_order_status,
        'process_return': process_return,
        'apply_discount': apply_discount,
        'get_product_recommendations': get_product_recommendations
    }
    
    if function_name in function_map:
        return function_map[function_name](arguments)
    else:
        return jsonify({"error": "Function not found"}), 404

def check_order_status(args):
    order_id = args.get('order_id')
    
    # Query your database
    conn = sqlite3.connect('orders.db')
    cursor = conn.cursor()
    
    cursor.execute("""
        SELECT status, tracking_number, estimated_delivery, total_amount
        FROM orders WHERE order_id = ?
    """, (order_id,))
    
    result = cursor.fetchone()
    conn.close()
    
    if result:
        return jsonify({
            "success": True,
            "data": {
                "order_id": order_id,
                "status": result[0],
                "tracking_number": result[1],
                "estimated_delivery": result[2],
                "total_amount": result[3]
            }
        })
    else:
        return jsonify({
            "success": False,
            "error": f"Order {order_id} not found"
        })

def process_return(args):
    order_id = args.get('order_id')
    reason = args.get('reason')
    
    # Process return logic
    return_id = f"RET-{datetime.now().strftime('%Y%m%d%H%M%S')}"
    
    # Insert into returns table
    conn = sqlite3.connect('orders.db')
    cursor = conn.cursor()
    
    cursor.execute("""
        INSERT INTO returns (return_id, order_id, reason, status, created_at)
        VALUES (?, ?, ?, 'pending', ?)
    """, (return_id, order_id, reason, datetime.now()))
    
    conn.commit()
    conn.close()
    
    return jsonify({
        "success": True,
        "data": {
            "return_id": return_id,
            "status": "Return initiated successfully",
            "next_steps": "You'll receive return instructions via email within 24 hours"
        }
    })

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

CRM Integration

import requests
from flask import Flask, request, jsonify

app = Flask(__name__)

# Your CRM API credentials
CRM_API_KEY = "your_crm_api_key"
CRM_BASE_URL = "https://api.yourcrm.com/v1"

@app.route('/tools', methods=['GET'])
def list_tools():
    return jsonify({
        "tools": [
            {
                "name": "get_customer_profile",
                "description": "Retrieve complete customer profile from CRM",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "customer_id": {"type": "string"},
                        "email": {"type": "string"}
                    }
                }
            },
            {
                "name": "create_support_ticket",
                "description": "Create a new support ticket in CRM",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "customer_id": {"type": "string"},
                        "title": {"type": "string"},
                        "description": {"type": "string"},
                        "priority": {"type": "string", "enum": ["low", "medium", "high", "urgent"]}
                    },
                    "required": ["customer_id", "title", "description"]
                }
            },
            {
                "name": "update_customer_notes",
                "description": "Add notes to customer profile",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "customer_id": {"type": "string"},
                        "note": {"type": "string"},
                        "note_type": {"type": "string", "enum": ["general", "support", "sales"]}
                    },
                    "required": ["customer_id", "note"]
                }
            },
            {
                "name": "schedule_callback",
                "description": "Schedule a callback for the customer",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "customer_id": {"type": "string"},
                        "preferred_time": {"type": "string"},
                        "reason": {"type": "string"}
                    },
                    "required": ["customer_id", "preferred_time"]
                }
            }
        ]
    })

@app.route('/call', methods=['POST'])
def call_function():
    data = request.get_json()
    function_name = data.get('name')
    arguments = data.get('arguments', {})
    
    try:
        if function_name == 'get_customer_profile':
            return get_customer_profile(arguments)
        elif function_name == 'create_support_ticket':
            return create_support_ticket(arguments)
        elif function_name == 'update_customer_notes':
            return update_customer_notes(arguments)
        elif function_name == 'schedule_callback':
            return schedule_callback(arguments)
        else:
            return jsonify({"error": "Function not found"}), 404
    except Exception as e:
        return jsonify({"error": str(e)}), 500

def get_customer_profile(args):
    customer_id = args.get('customer_id')
    email = args.get('email')
    
    # Query CRM API
    if customer_id:
        url = f"{CRM_BASE_URL}/customers/{customer_id}"
    elif email:
        url = f"{CRM_BASE_URL}/customers/search?email={email}"
    else:
        return jsonify({"error": "customer_id or email required"})
    
    response = requests.get(url, headers={
        "Authorization": f"Bearer {CRM_API_KEY}"
    })
    
    if response.status_code == 200:
        customer_data = response.json()
        return jsonify({
            "success": True,
            "data": {
                "customer_id": customer_data.get('id'),
                "name": customer_data.get('name'),
                "email": customer_data.get('email'),
                "phone": customer_data.get('phone'),
                "tier": customer_data.get('tier'),
                "total_spent": customer_data.get('total_spent'),
                "last_purchase": customer_data.get('last_purchase_date'),
                "open_tickets": customer_data.get('open_tickets_count', 0)
            }
        })
    else:
        return jsonify({
            "success": False,
            "error": "Customer not found in CRM"
        })

def create_support_ticket(args):
    ticket_data = {
        "customer_id": args.get('customer_id'),
        "title": args.get('title'),
        "description": args.get('description'),
        "priority": args.get('priority', 'medium'),
        "source": "ai_assistant"
    }
    
    response = requests.post(
        f"{CRM_BASE_URL}/tickets",
        json=ticket_data,
        headers={"Authorization": f"Bearer {CRM_API_KEY}"}
    )
    
    if response.status_code == 201:
        ticket = response.json()
        return jsonify({
            "success": True,
            "data": {
                "ticket_id": ticket.get('id'),
                "ticket_number": ticket.get('number'),
                "status": ticket.get('status'),
                "assigned_to": ticket.get('assigned_agent'),
                "message": "Support ticket created successfully"
            }
        })
    else:
        return jsonify({
            "success": False,
            "error": "Failed to create ticket"
        })

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

Using MCP Functions in Prompts

Once your MCP is connected, reference functions in your assistant prompts:

You are a customer service specialist with access to our order management system.

Available functions:
- {{function.check_order_status(order_id)}} - Get order status and tracking
- {{function.process_return(order_id, reason)}} - Initiate returns
- {{function.apply_discount(customer_id, code)}} - Apply discount codes
- {{function.get_customer_profile(customer_id)}} - Get customer details

When customer asks about their order:
1. Ask for order number if not provided
2. Call {{function.check_order_status(order_id)}}
3. Provide helpful summary of the results

For return requests:
1. Get order details first
2. Confirm return eligibility
3. Call {{function.process_return(order_id, reason)}}
4. Explain next steps to customer

Always validate order numbers and handle errors gracefully.

Testing Your MCP

Test Server Locally

Verify your MCP server responds correctly:

# Test tools endpoint
curl http://localhost:5000/tools

# Test function call
curl -X POST http://localhost:5000/call \
  -H "Content-Type: application/json" \
  -d '{
    "name": "check_order_status",
    "arguments": {"order_id": "ORD-001"}
  }'

Deploy to Production

Deploy your MCP server to a public URL:

Use platforms like Heroku, Railway, or AWS
Ensure HTTPS is enabled
Set up proper authentication

Test with Plati

Send test messages to verify integration:

curl -X POST https://api.plati.ai/conversation/test_user/chat/{channel_id} \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -d '{"text": "Check my order ORD-001"}'

Security Best Practices

Authentication

Always authenticate MCP requests:

@app.before_request
def authenticate():
    if request.endpoint in ['call', 'tools']:
        auth_header = request.headers.get('Authorization')
        if not auth_header or not verify_token(auth_header):
            return jsonify({"error": "Unauthorized"}), 401

Input Validation

Validate all function inputs:

def validate_order_id(order_id):
    if not order_id or not re.match(r'^ORD-\d+$', order_id):
        raise ValueError("Invalid order ID format")

Rate Limiting

Implement rate limiting:

from flask_limiter import Limiter

limiter = Limiter(
    app,
    key_func=lambda: request.headers.get('Authorization'),
    default_limits=["100 per minute"]
)

Error Handling

Handle errors gracefully:

try:
    result = external_api_call()
    return jsonify({"success": True, "data": result})
except ExternalAPIError as e:
    return jsonify({
        "success": False, 
        "error": "Service temporarily unavailable"
    })

Advanced MCP Patterns

Stateful Functions

Maintain state across function calls:

# Store conversation context
conversation_context = {}

def get_order_history(args):
    customer_id = args.get('customer_id')
    
    # Store context for follow-up questions
    conversation_context[customer_id] = {
        'last_query': 'order_history',
        'orders': fetch_orders(customer_id)
    }
    
    return jsonify({
        "success": True,
        "data": conversation_context[customer_id]['orders']
    })

Async Functions

Handle long-running operations:

import asyncio
from threading import Thread

@app.route('/call', methods=['POST'])
def call_function():
    data = request.get_json()
    
    if data.get('name') == 'generate_report':
        # Start async task
        task_id = start_report_generation(data.get('arguments'))
        return jsonify({
            "success": True,
            "task_id": task_id,
            "message": "Report generation started. You'll be notified when complete."
        })

Function Chaining

Chain multiple functions together:

def process_order_issue(args):
    order_id = args.get('order_id')
    
    # Step 1: Get order details
    order = get_order_details(order_id)
    
    # Step 2: Check if refund eligible
    if order['status'] in ['delivered', 'completed']:
        # Step 3: Process refund
        refund_result = process_refund(order_id)
        return jsonify({
            "success": True,
            "actions_taken": [
                "Retrieved order details",
                "Verified refund eligibility", 
                "Processed refund"
            ],
            "refund_id": refund_result['refund_id']
        })

MCP SDK (Coming Soon)

We’re developing an official SDK to simplify MCP development:

from plati_mcp import MCP, Function

mcp = MCP(name="Order Management")

@mcp.function(
    name="check_order_status",
    description="Check order status and tracking",
    parameters={
        "order_id": {"type": "string", "required": True}
    }
)
def check_order_status(order_id: str):
    # Your implementation
    return {"status": "shipped", "tracking": "123456"}

@mcp.function(
    name="process_return", 
    description="Process a return request",
    parameters={
        "order_id": {"type": "string", "required": True},
        "reason": {"type": "string", "required": True}
    }
)
def process_return(order_id: str, reason: str):
    # Your implementation
    return {"return_id": "RET-123", "status": "approved"}

if __name__ == '__main__':
    mcp.run(host='0.0.0.0', port=5000)

Next Steps

Prompt Engineering

Learn to use MCP functions in dynamic prompts

Contact Support

Connect MCP functions with user data

Scheduled Tasks

Combine MCP with automated workflows

Troubleshooting

Debug MCP integration issues

Getting Started

Advanced

API Reference

Mcp

Model Context Protocol (MCP)

Understanding MCP

What MCP Enables

How It Works

MCP Architecture Flow

Building Your First MCP Server

1. Basic HTTP Server

2. Register Your MCP in Plati

3. Connect MCP to Assistant

Real-World MCP Examples

E-commerce Integration

CRM Integration

Using MCP Functions in Prompts

Testing Your MCP

Security Best Practices

Advanced MCP Patterns

Stateful Functions

Async Functions

Function Chaining

MCP SDK (Coming Soon)

Next Steps

Prompt Engineering

Contact Support

Scheduled Tasks

Troubleshooting

Getting Started

Advanced

API Reference

Documentation Index

​Model Context Protocol (MCP)

​Understanding MCP

What MCP Enables

How It Works

​MCP Architecture Flow

​Building Your First MCP Server

​1. Basic HTTP Server

​2. Register Your MCP in Plati

​3. Connect MCP to Assistant

​Real-World MCP Examples

​E-commerce Integration

​CRM Integration

​Using MCP Functions in Prompts

​Testing Your MCP

​Security Best Practices

​Advanced MCP Patterns

​Stateful Functions

​Async Functions

​Function Chaining

​MCP SDK (Coming Soon)

​Next Steps

Prompt Engineering

Contact Support

Scheduled Tasks

Troubleshooting

Model Context Protocol (MCP)

Understanding MCP

MCP Architecture Flow

Building Your First MCP Server

1. Basic HTTP Server

2. Register Your MCP in Plati

3. Connect MCP to Assistant

Real-World MCP Examples

E-commerce Integration

CRM Integration

Using MCP Functions in Prompts

Testing Your MCP

Security Best Practices

Advanced MCP Patterns

Stateful Functions

Async Functions

Function Chaining

MCP SDK (Coming Soon)

Next Steps