Connecting MCP to Your Business APIs

Overview

Connecting MCP to your business APIs transforms static AI models into dynamic agents that can interact with your existing systems. This guide covers practical patterns for integrating with REST APIs, GraphQL endpoints, and common authentication schemes used in business applications.

Business Value: Instead of manually copying data between systems or writing custom integration scripts, MCP enables AI to directly access and manipulate your business data through existing APIs.

Common Business API Integration Patterns

CRM Systems

• Salesforce REST API: Lead management, opportunity tracking
• HubSpot API: Contact sync, deal pipeline updates
• Pipedrive API: Sales activity automation

Financial Systems

• QuickBooks API: Invoice generation, expense tracking
• Stripe API: Payment processing, subscription management
• Xero API: Financial reporting, tax calculations

Communication Platforms

• Slack API: Message posting, channel management
• Microsoft Teams API: Team collaboration, file sharing
• Twilio API: SMS notifications, call management

Setting Up API Integration with MCP

Basic REST API Integration

# api_server.py
import requests
from mcp.server.fastmcp import FastMCP
from typing import Optional, Dict, Any
import os

mcp = FastMCP("BusinessAPIServer")

class APIClient:
    def __init__(self, base_url: str, api_key: str):
        self.base_url = base_url.rstrip('/')
        self.session = requests.Session()
        self.session.headers.update({
            'Authorization': f'Bearer {api_key}',
            'Content-Type': 'application/json'
        })
    
    def make_request(self, method: str, endpoint: str, **kwargs) -> Dict[Any, Any]:
        """Make authenticated API request with error handling"""
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        
        try:
            response = self.session.request(method, url, **kwargs)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            return {"error": str(e), "status_code": getattr(e.response, 'status_code', None)}

# Initialize API client (customize for your API)
api_client = APIClient(
    base_url=os.getenv('API_BASE_URL', 'https://api.yourbusiness.com'),
    api_key=os.getenv('API_KEY', 'your-api-key')
)

@mcp.tool()
def get_customer_data(customer_id: str) -> Dict[Any, Any]:
    """Retrieve customer information from business API"""
    return api_client.make_request('GET', f'/customers/{customer_id}')

@mcp.tool()
def create_customer(name: str, email: str, phone: Optional[str] = None) -> Dict[Any, Any]:
    """Create new customer record"""
    data = {"name": name, "email": email}
    if phone:
        data["phone"] = phone
    
    return api_client.make_request('POST', '/customers', json=data)

@mcp.tool()
def update_customer(customer_id: str, **updates) -> Dict[Any, Any]:
    """Update existing customer information"""
    return api_client.make_request('PATCH', f'/customers/{customer_id}', json=updates)

if __name__ == "__main__":
    mcp.run()

Advanced Authentication Patterns

OAuth 2.0 Integration

# oauth_integration.py
import requests
from datetime import datetime, timedelta
from mcp.server.fastmcp import FastMCP
import json
import os

class OAuthAPIClient:
    def __init__(self, client_id: str, client_secret: str, token_url: str, base_url: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_url = token_url
        self.base_url = base_url
        self.access_token = None
        self.token_expires = None
    
    def get_access_token(self):
        """Obtain OAuth access token"""
        if self.access_token and self.token_expires > datetime.now():
            return self.access_token
        
        response = requests.post(self.token_url, data={
            'grant_type': 'client_credentials',
            'client_id': self.client_id,
            'client_secret': self.client_secret
        })
        
        if response.status_code == 200:
            token_data = response.json()
            self.access_token = token_data['access_token']
            # Set expiry time with buffer
            expires_in = token_data.get('expires_in', 3600) - 300
            self.token_expires = datetime.now() + timedelta(seconds=expires_in)
            return self.access_token
        else:
            raise Exception(f"Failed to obtain access token: {response.text}")
    
    def make_authenticated_request(self, method: str, endpoint: str, **kwargs):
        """Make API request with automatic token refresh"""
        token = self.get_access_token()
        headers = kwargs.get('headers', {})
        headers['Authorization'] = f'Bearer {token}'
        kwargs['headers'] = headers
        
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        response = requests.request(method, url, **kwargs)
        
        return response.json() if response.status_code < 400 else {
            "error": response.text,
            "status_code": response.status_code
        }

# Example: Salesforce integration
oauth_client = OAuthAPIClient(
    client_id=os.getenv('SALESFORCE_CLIENT_ID'),
    client_secret=os.getenv('SALESFORCE_CLIENT_SECRET'),
    token_url='https://login.salesforce.com/services/oauth2/token',
    base_url='https://your-instance.salesforce.com'
)

mcp = FastMCP("SalesforceIntegration")

@mcp.tool()
def get_leads(limit: int = 10) -> Dict[Any, Any]:
    """Retrieve recent leads from Salesforce"""
    query = f"SELECT Id, Name, Email, Company FROM Lead ORDER BY CreatedDate DESC LIMIT {limit}"
    return oauth_client.make_authenticated_request(
        'GET', 
        f'/services/data/v58.0/query',
        params={'q': query}
    )

@mcp.tool()
def create_lead(name: str, email: str, company: str) -> Dict[Any, Any]:
    """Create new lead in Salesforce"""
    data = {
        "Name": name,
        "Email": email,
        "Company": company,
        "LeadSource": "MCP Integration"
    }
    return oauth_client.make_authenticated_request(
        'POST',
        '/services/data/v58.0/sobjects/Lead',
        json=data
    )

GraphQL Integration

# graphql_integration.py
import requests
from mcp.server.fastmcp import FastMCP
from typing import Dict, Any, Optional

class GraphQLClient:
    def __init__(self, endpoint: str, headers: Optional[Dict[str, str]] = None):
        self.endpoint = endpoint
        self.headers = headers or {}
    
    def execute_query(self, query: str, variables: Optional[Dict[str, Any]] = None) -> Dict[Any, Any]:
        """Execute GraphQL query"""
        payload = {"query": query}
        if variables:
            payload["variables"] = variables
        
        response = requests.post(
            self.endpoint,
            json=payload,
            headers=self.headers
        )
        
        return response.json()

# Example: GitHub GraphQL API
github_client = GraphQLClient(
    endpoint="https://api.github.com/graphql",
    headers={"Authorization": f"Bearer {os.getenv('GITHUB_TOKEN')}"}
)

mcp = FastMCP("GitHubGraphQLServer")

@mcp.tool()
def get_repository_info(owner: str, name: str) -> Dict[Any, Any]:
    """Get repository information using GitHub GraphQL API"""
    query = """
    query($owner: String!, $name: String!) {
        repository(owner: $owner, name: $name) {
            name
            description
            stargazerCount
            forkCount
            primaryLanguage {
                name
            }
            issues(states: OPEN) {
                totalCount
            }
            pullRequests(states: OPEN) {
                totalCount
            }
        }
    }
    """
    
    variables = {"owner": owner, "name": name}
    return github_client.execute_query(query, variables)

@mcp.tool()
def get_user_repositories(username: str, first: int = 10) -> Dict[Any, Any]:
    """Get user's repositories"""
    query = """
    query($username: String!, $first: Int!) {
        user(login: $username) {
            repositories(first: $first, orderBy: {field: UPDATED_AT, direction: DESC}) {
                nodes {
                    name
                    description
                    url
                    stargazerCount
                    primaryLanguage {
                        name
                    }
                }
            }
        }
    }
    """
    
    variables = {"username": username, "first": first}
    return github_client.execute_query(query, variables)

Rate Limiting and Error Handling

Implementing Rate Limiting

# rate_limited_client.py
import time
from datetime import datetime, timedelta
from collections import deque
from typing import Dict, Any

class RateLimitedAPIClient:
    def __init__(self, base_url: str, api_key: str, rate_limit: int = 100, window_seconds: int = 60):
        self.base_url = base_url
        self.api_key = api_key
        self.rate_limit = rate_limit
        self.window_seconds = window_seconds
        self.request_times = deque()
    
    def wait_if_needed(self):
        """Implement sliding window rate limiting"""
        now = datetime.now()
        cutoff = now - timedelta(seconds=self.window_seconds)
        
        # Remove old requests outside the window
        while self.request_times and self.request_times[0] < cutoff:
            self.request_times.popleft()
        
        # If we've hit the rate limit, wait
        if len(self.request_times) >= self.rate_limit:
            sleep_time = (self.request_times[0] + timedelta(seconds=self.window_seconds) - now).total_seconds()
            if sleep_time > 0:
                time.sleep(sleep_time)
        
        self.request_times.append(now)
    
    def make_request(self, method: str, endpoint: str, retry_count: int = 3, **kwargs) -> Dict[Any, Any]:
        """Make rate-limited request with retries"""
        for attempt in range(retry_count):
            try:
                self.wait_if_needed()
                
                response = requests.request(
                    method,
                    f"{self.base_url}/{endpoint.lstrip('/')}",
                    headers={"Authorization": f"Bearer {self.api_key}"},
                    **kwargs
                )
                
                if response.status_code == 429:  # Rate limited
                    retry_after = int(response.headers.get('Retry-After', 60))
                    time.sleep(retry_after)
                    continue
                
                response.raise_for_status()
                return response.json()
                
            except requests.exceptions.RequestException as e:
                if attempt == retry_count - 1:  # Last attempt
                    return {"error": str(e), "attempt": attempt + 1}
                time.sleep(2 ** attempt)  # Exponential backoff
        
        return {"error": "Max retries exceeded"}

Environment Configuration

Create a .env file for your API credentials:

# .env
API_BASE_URL=https://api.yourbusiness.com
API_KEY=your-api-key-here

# Salesforce OAuth
SALESFORCE_CLIENT_ID=your-salesforce-client-id
SALESFORCE_CLIENT_SECRET=your-salesforce-client-secret

# GitHub
GITHUB_TOKEN=your-github-token

# Other services
SLACK_BOT_TOKEN=xoxb-your-slack-token
STRIPE_SECRET_KEY=sk_test_your-stripe-key

Load environment variables in your MCP server:

# Load environment variables
from dotenv import load_dotenv
load_dotenv()

Business Use Case Examples

Customer Service Integration

@mcp.tool()
def create_support_ticket(customer_email: str, subject: str, description: str, priority: str = "medium") -> Dict[Any, Any]:
    """Create support ticket in business system"""
    ticket_data = {
        "customer_email": customer_email,
        "subject": subject,
        "description": description,
        "priority": priority,
        "status": "open",
        "created_at": datetime.now().isoformat()
    }
    
    # Create in helpdesk system
    result = api_client.make_request('POST', '/tickets', json=ticket_data)
    
    # Send notification via Slack
    if 'id' in result:
        slack_message = f"New support ticket #{result['id']}: {subject} (Priority: {priority})"
        send_slack_notification("#support", slack_message)
    
    return result

def send_slack_notification(channel: str, message: str):
    """Send notification to Slack channel"""
    slack_client.make_request('POST', 'https://slack.com/api/chat.postMessage', json={
        "channel": channel,
        "text": message
    })

Sales Pipeline Automation

@mcp.tool()
def update_deal_stage(deal_id: str, stage: str, notes: Optional[str] = None) -> Dict[Any, Any]:
    """Update deal stage and log activity"""
    update_data = {"stage": stage}
    if notes:
        update_data["notes"] = notes
    
    # Update in CRM
    result = crm_client.make_request('PATCH', f'/deals/{deal_id}', json=update_data)
    
    # Log activity
    if 'id' in result:
        activity_data = {
            "deal_id": deal_id,
            "activity_type": "stage_change",
            "description": f"Deal moved to {stage}",
            "notes": notes
        }
        crm_client.make_request('POST', '/activities', json=activity_data)
    
    return result

Testing Your API Integration

# test_api_integration.py
def test_api_connection():
    """Test basic API connectivity"""
    try:
        response = api_client.make_request('GET', '/health')
        print("✅ API connection successful")
        return True
    except Exception as e:
        print(f"❌ API connection failed: {e}")
        return False

def test_authentication():
    """Test API authentication"""
    try:
        response = api_client.make_request('GET', '/user/profile')
        if 'error' not in response:
            print("✅ Authentication successful")
            return True
        else:
            print(f"❌ Authentication failed: {response['error']}")
            return False
    except Exception as e:
        print(f"❌ Authentication error: {e}")
        return False

if __name__ == "__main__":
    print("Testing API Integration...")
    test_api_connection()
    test_authentication()

Security Best Practices

1. Environment Variables: Store all API credentials in environment variables, never in code
2. Token Rotation: Implement automatic token refresh for OAuth flows
3. Request Logging: Log API requests for debugging but exclude sensitive data
4. Rate Limiting: Respect API rate limits to avoid service disruption
5. Error Handling: Implement comprehensive error handling and user-friendly error messages

Common Business Integration Patterns

Multi-System Data Sync

@mcp.tool()
def sync_customer_data(customer_id: str) -> Dict[str, Any]:
    """Sync customer data across multiple business systems"""
    results = {}
    
    # Get customer from CRM
    crm_data = crm_client.make_request('GET', f'/customers/{customer_id}')
    results['crm'] = crm_data
    
    # Update in billing system
    if 'email' in crm_data:
        billing_data = {
            "email": crm_data['email'],
            "name": crm_data['name'],
            "updated_from": "crm_sync"
        }
        results['billing'] = billing_client.make_request(
            'PUT', f'/customers/{customer_id}', json=billing_data
        )
    
    # Update in support system
    support_data = {
        "customer_id": customer_id,
        "priority_level": crm_data.get('tier', 'standard')
    }
    results['support'] = support_client.make_request(
        'PUT', f'/customers/{customer_id}', json=support_data
    )
    
    return results

Next Steps

Once you’ve mastered basic API integration, explore:

• Business Document Processing - Handle PDFs, contracts, and reports
• Business Tool Integration - Connect project management and communication tools
• Salesforce MCP Integration - Complete CRM implementation
• MCP Security for Enterprise - Production security patterns

Ready to implement? Start with your most frequently used business API and gradually expand to additional systems for maximum efficiency gains.