Keap MCP Server

A high-performance Model Context Protocol (MCP) server for interacting with Keap CRM data with advanced features including HTTP/2 support, comprehensive diagnostics, and bulk operations.

Features

Core Contact & Tag Management

• Comprehensive Contact Management - List, search, filter, and get detailed contact information

• Advanced Tag Operations - Full tag lifecycle management including creation, querying, and batch operations

• Batch Tag Operations - Apply or remove multiple tags from multiple contacts efficiently

• Custom Field Operations - Search contacts by custom field values and bulk update custom fields

• Complex Logical Filtering - Support for AND, OR, NOT operators with nested conditions

Performance & Optimization

• HTTP/2 Support - Enhanced connection performance with connection pooling

• Intelligent Rate Limiting - Daily request limits with adaptive backoff strategies

• Query Optimization - Intelligent server-side vs client-side filtering with performance analytics

• Performance Monitoring - Real-time query analysis and optimization suggestions

• Persistent Caching - SQLite-based caching to reduce API calls and improve performance

Advanced Features

• Comprehensive Diagnostics - API performance metrics, system monitoring, and health checks

• Enhanced Error Handling - Robust retry logic with exponential backoff for different error types

• Bulk Custom Field Updates - Efficiently set custom field values across multiple contacts

• Advanced Filter Operators - 15+ operators including BETWEEN, IN, SINCE, STARTS_WITH, etc.

• ID List Operations - Utility functions for working with contact and tag ID sets

Related MCP server: Quick-start Auto MCP

Architecture

The Keap MCP Server uses a streamlined, high-performance architecture:

• API Client (src/api/client.py) - Enhanced Keap API communication with HTTP/2, rate limiting, and diagnostics

• MCP Tools (src/mcp/) - Comprehensive MCP protocol implementation with optimization

• Cache Manager (src/cache/) - SQLite-based persistent caching with intelligent invalidation

• Optimization Engine (src/mcp/optimization/) - Query optimization and performance analytics

• Schemas (src/schemas/) - Data validation and models

• Utils (src/utils/) - Shared utilities for contact processing and filtering

MCP Tools

The server exposes 17 comprehensive MCP tools:

Contact Operations

1. list_contacts - List contacts with filtering and pagination (now optimized)

2. search_contacts_by_email - Find contacts by email address

3. search_contacts_by_name - Find contacts by name

4. get_contact_details - Get detailed information about a specific contact

5. query_contacts_by_custom_field - Query contacts by custom field value

Tag Operations

6. get_tags - Retrieve tags with optional filtering

7. get_tag_details - Get detailed information about a specific tag

8. get_contacts_with_tag - Get contacts that have a specific tag

9. create_tag - Create a new tag

Tag Management (Batch Operations)

10. modify_tags - Add or remove tags from contacts

11. apply_tags_to_contacts - Apply multiple tags to multiple contacts using batch operations

12. remove_tags_from_contacts - Remove multiple tags from multiple contacts

Custom Field Management

13. set_custom_field_values - Bulk update custom field values across multiple contacts

Advanced Query & Performance Operations

14. query_contacts_optimized - Advanced contact query with optimization and performance analytics

15. analyze_query_performance - Analyze query performance and optimization potential

System Operations

16. get_api_diagnostics - Comprehensive API diagnostics and performance metrics

Utility Operations

17. intersect_id_lists - Find the intersection of multiple ID lists

Getting Started

Prerequisites

• Python 3.9 or higher

• Keap API credentials

Installation

1. Clone the repository:

   git clone https://github.com/yourusername/keapmcp.git
   cd keapmcp

2. Install dependencies:

   pip install -r requirements.txt

3. Configure your Keap API credentials:

   • The application uses a .env file for configuration

   • The API key has been copied from keapsync, but you can modify it if needed

   • The configuration includes:

     KEAP_API_KEY=your_api_key_here
     KEAP_API_BASE_URL=https://api.infusionsoft.com/crm/rest/v1
     KEAP_MCP_HOST=127.0.0.1
     KEAP_MCP_PORT=5000
     KEAP_MCP_LOG_LEVEL=INFO
     KEAP_MCP_LOG_FILE=keap_mcp_server.log
     KEAP_MCP_CACHE_ENABLED=true
     KEAP_MCP_CACHE_TTL=3600

Running the Server

python run.py --host 127.0.0.1 --port 5000

Command-line options:

• --host - Host to bind to (default: 127.0.0.1)

• --port - Port to listen on (default: 5000)

• --log-level - Logging level (default: INFO)

• --log-file - Log file path (default: keap_mcp_server.log)

• --no-console-log - Disable console logging

Testing & Coverage

The Keap MCP Server includes a comprehensive test suite to ensure reliability and correct operation with full CI/CD pipeline integration.

Quick Testing

Use the Makefile for easy test execution:

# Run all unit tests
make test

# Run tests with coverage reporting  
make coverage

# Generate HTML coverage report
make coverage-html

# Run service-specific tests
make test-services
make test-models

# Full development workflow
make dev-test

CI/CD Integration

The project includes comprehensive automated testing and quality assurance:

• Continuous Integration: Tests run on Python 3.9, 3.10, and 3.11

• Coverage Tracking: Minimum 70% unit test coverage enforced

• Code Quality: Ruff linting and formatting checks

• Security Scanning: Bandit security analysis and Safety dependency checks

• Type Checking: MyPy static type analysis (non-blocking)

• Pre-commit Hooks: Automated code quality checks on commit

• Build Verification: Import and initialization testing

CI Pipeline Jobs

1. Lint Code: Ruff linting and formatting validation

2. Unit Tests: Full test suite with coverage reporting across Python versions

3. Security Scan: Bandit and Safety security analysis

4. Type Check: MyPy static type checking

5. Build Test: Package import and server initialization verification

6. Coverage Report: HTML coverage reports for pull requests

Test Categories

• Unit Tests (tests/unit/) - Individual component testing with comprehensive mocking

• Integration Tests (tests/integration/) - End-to-end functionality verification

• Performance Tests (tests/performance/) - Load and optimization validation

• API Validation - Keap API response format verification

Coverage Requirements

• Current Coverage: 55% integration coverage, varies by component

• API Client: Core functionality tested with comprehensive mocking

• MCP Tools: Integration tests with mock dependencies

• Cache System: Comprehensive persistence and performance testing

• Utilities: Contact processing and filtering functionality tested

• Optimization: Performance analytics and query optimization covered

Running Specific Tests

# Run all tests
python -m pytest tests/ -v

# With coverage reporting
python -m pytest tests/ --cov=src --cov-fail-under=90

# Integration tests (requires running server)
python -m pytest tests/integration/ -v

Using the MCP Server

Example: List Contacts (Now Optimized)

{
  "function": "list_contacts",
  "params": {
    "filters": [
      { "field": "email", "operator": "contains", "value": "@company.com" }
    ],
    "limit": 50,
    "include": ["id", "given_name", "family_name", "email"]
  }
}

Note: list_contacts now uses the optimization engine internally for better performance. For detailed performance metrics, use query_contacts_optimized directly.

Example: Search by Email

{
  "function": "search_contacts_by_email",
  "params": {
    "email": "john.doe@company.com",
    "include": ["id", "given_name", "family_name", "email", "tags"]
  }
}

Example: Get Tags

{
  "function": "get_tags",
  "params": {
    "include_categories": true,
    "limit": 100
  }
}

Example: Batch Tag Operations

{
  "function": "apply_tags_to_contacts",
  "params": {
    "tag_ids": ["123", "456"],
    "contact_ids": ["1001", "1002", "1003"]
  }
}

Example: Custom Field Query

{
  "function": "query_contacts_by_custom_field",
  "params": {
    "field_id": "7",
    "field_value": "Engineering",
    "operator": "contains",
    "include": ["id", "given_name", "family_name", "email"]
  }
}

Example: Create New Tag

{
  "function": "create_tag",
  "params": {
    "name": "VIP Customer",
    "description": "High-value customer segment",
    "category_id": "2"
  }
}

Example: Bulk Custom Field Updates

{
  "function": "set_custom_field_values",
  "params": {
    "field_id": "7",
    "contact_ids": ["1001", "1002", "1003"],
    "common_value": "VIP Customer"
  }
}

Or with individual values per contact:

{
  "function": "set_custom_field_values",
  "params": {
    "field_id": "7",
    "contact_values": {
      "1001": "Gold Tier",
      "1002": "Silver Tier", 
      "1003": "Bronze Tier"
    }
  }
}

Example: API Diagnostics

{
  "function": "get_api_diagnostics",
  "params": {}
}

Example: ID List Intersection

{
  "function": "intersect_id_lists",
  "params": {
    "lists": [
      {"list_id": "active_contacts", "item_ids": ["1", "2", "3", "4"]},
      {"list_id": "newsletter_subscribers", "item_ids": ["2", "3", "5", "6"]}
    ],
    "id_field": "item_ids"
  }
}

Performance Features

HTTP/2 Support

The server uses HTTP/2 for enhanced performance with connection pooling and keepalive connections.

Rate Limiting

• Daily request limits (25,000 requests/day by default)

• Intelligent backoff strategies

• Rate limit monitoring and diagnostics

Caching Strategy

• SQLite-based persistent caching

• Intelligent cache invalidation

• TTL-based expiration

• Cache hit/miss tracking

Error Handling

• Exponential backoff for retries

• Different strategies for timeout, network, and HTTP errors

• Comprehensive error tracking and diagnostics

License

This project is licensed under the MIT License - see the LICENSE file for details.