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
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
list_contacts- List contacts with filtering and pagination (now optimized)search_contacts_by_email- Find contacts by email addresssearch_contacts_by_name- Find contacts by nameget_contact_details- Get detailed information about a specific contactquery_contacts_by_custom_field- Query contacts by custom field value
Tag Operations
get_tags- Retrieve tags with optional filteringget_tag_details- Get detailed information about a specific tagget_contacts_with_tag- Get contacts that have a specific tagcreate_tag- Create a new tag
Tag Management (Batch Operations)
modify_tags- Add or remove tags from contactsapply_tags_to_contacts- Apply multiple tags to multiple contacts using batch operationsremove_tags_from_contacts- Remove multiple tags from multiple contacts
Custom Field Management
set_custom_field_values- Bulk update custom field values across multiple contacts
Advanced Query & Performance Operations
query_contacts_optimized- Advanced contact query with optimization and performance analyticsanalyze_query_performance- Analyze query performance and optimization potential
System Operations
get_api_diagnostics- Comprehensive API diagnostics and performance metrics
Utility Operations
intersect_id_lists- Find the intersection of multiple ID lists
Getting Started
Prerequisites
- Python 3.9 or higher
- Keap API credentials
Installation
-
Clone the repository:
git clone https://github.com/yourusername/keapmcp.git cd keapmcp -
Install dependencies:
pip install -r requirements.txt -
Configure your Keap API credentials:
- The application uses a
.envfile 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
- The application uses a
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
- Lint Code: Ruff linting and formatting validation
- Unit Tests: Full test suite with coverage reporting across Python versions
- Security Scan: Bandit and Safety security analysis
- Type Check: MyPy static type checking
- Build Test: Package import and server initialization verification
- 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.