claude-safe-mcp

# Keap MCP Server [](https://github.com/yourusername/keapmcp/actions) [](https://github.com/yourusername/keapmcp) [](LICENSE) 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 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: ```bash # 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 automated testing and coverage reporting: - **Continuous Integration**: Tests run on Python 3.9, 3.10, and 3.11 - **Coverage Tracking**: Minimum 60% coverage enforced - **Pre-commit Checks**: Code formatting, linting, and security checks - **Coverage Badges**: Automatically updated on main branch ### 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 ```bash # 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) ```json { "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 ```json { "function": "search_contacts_by_email", "params": { "email": "john.doe@company.com", "include": ["id", "given_name", "family_name", "email", "tags"] } } ``` ### Example: Get Tags ```json { "function": "get_tags", "params": { "include_categories": true, "limit": 100 } } ``` ### Example: Batch Tag Operations ```json { "function": "apply_tags_to_contacts", "params": { "tag_ids": ["123", "456"], "contact_ids": ["1001", "1002", "1003"] } } ``` ### Example: Custom Field Query ```json { "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 ```json { "function": "create_tag", "params": { "name": "VIP Customer", "description": "High-value customer segment", "category_id": "2" } } ``` ### Example: Bulk Custom Field Updates ```json { "function": "set_custom_field_values", "params": { "field_id": "7", "contact_ids": ["1001", "1002", "1003"], "common_value": "VIP Customer" } } ``` Or with individual values per contact: ```json { "function": "set_custom_field_values", "params": { "field_id": "7", "contact_values": { "1001": "Gold Tier", "1002": "Silver Tier", "1003": "Bronze Tier" } } } ``` ### Example: API Diagnostics ```json { "function": "get_api_diagnostics", "params": {} } ``` ### Example: ID List Intersection ```json { "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.