test_server.py•6.05 kB
#!/usr/bin/env python3
"""
Test script for Drip MCP Server
Run this to verify your setup is working correctly.
"""
import os
import sys
import asyncio
import json
from pathlib import Path
# Add project root to path to allow imports
project_root = Path(__file__).parent.parent
sys.path.insert(0, str(project_root / "src"))
# Check for required environment variables
def check_environment():
"""Check if required environment variables are set."""
api_key = os.environ.get("DRIP_API_KEY")
account_id = os.environ.get("DRIP_ACCOUNT_ID")
if not api_key or not account_id:
print("❌ Environment variables not set!")
print("\nPlease set the following environment variables:")
print(" export DRIP_API_KEY='your-api-key'")
print(" export DRIP_ACCOUNT_ID='your-account-id'")
print("\nFind these in your Drip account:")
print(" - API Key: Settings → My User Settings → API Token")
print(" - Account ID: The number in your Drip dashboard URL")
return False
print("✅ Environment variables found")
print(f" - Account ID: {account_id}")
print(f" - API Key: {'*' * 10}{api_key[-4:]}") # Show last 4 chars only
return True
async def test_basic_functionality():
"""Test basic server functionality."""
try:
# Import the server module
from drip_mcp import (
list_tags,
list_custom_fields,
list_event_actions,
ListSubscribersInput,
list_subscribers
)
print("\n✅ Server module imported successfully")
# Test 1: List tags (read-only, safe operation)
print("\n📋 Testing: List Tags...")
try:
tags_result = await list_tags()
if "Tags in Account" in tags_result or "tags" in tags_result:
print(" ✅ List tags successful")
else:
print(f" ⚠️ Unexpected response: {tags_result[:100]}")
except Exception as e:
print(f" ❌ Error listing tags: {e}")
# Test 2: List custom fields (read-only, safe operation)
print("\n📋 Testing: List Custom Fields...")
try:
fields_result = await list_custom_fields()
if "Custom Field" in fields_result or "custom_field" in fields_result:
print(" ✅ List custom fields successful")
else:
print(f" ⚠️ Unexpected response: {fields_result[:100]}")
except Exception as e:
print(f" ❌ Error listing custom fields: {e}")
# Test 3: List subscribers with limit (read-only, safe operation)
print("\n📋 Testing: List Subscribers (limit 1)...")
try:
params = ListSubscribersInput(
status="active",
per_page=1,
response_format="json"
)
subscribers_result = await list_subscribers(params)
result_data = json.loads(subscribers_result)
if "subscribers" in result_data:
print(" ✅ List subscribers successful")
if result_data.get("meta"):
print(f" Total subscribers: {result_data['meta'].get('total_count', 'Unknown')}")
else:
print(f" ⚠️ Unexpected response structure")
except Exception as e:
print(f" ❌ Error listing subscribers: {e}")
# Test 4: List event actions (read-only, safe operation)
print("\n📋 Testing: List Event Actions...")
try:
events_result = await list_event_actions(per_page=5)
if "Event Actions" in events_result or "event_actions" in events_result:
print(" ✅ List event actions successful")
else:
print(f" ⚠️ Unexpected response: {events_result[:100]}")
except Exception as e:
print(f" ❌ Error listing event actions: {e}")
print("\n" + "="*50)
print("✅ Basic functionality tests completed!")
print("\nYour Drip MCP server is ready to use with Claude Code.")
print("\nNext steps:")
print("1. Add the server to your Claude Code configuration")
print("2. Restart Claude Code")
print("3. Start using natural language commands to interact with Drip")
except ImportError as e:
print(f"\n❌ Failed to import server module: {e}")
print("\nMake sure you're running this from the project root directory")
print("Run: python -m pytest tests/test_server.py")
return False
except Exception as e:
print(f"\n❌ Unexpected error: {e}")
return False
return True
def main():
"""Main test function."""
print("="*50)
print("Drip MCP Server Test Script")
print("="*50)
# Check environment
if not check_environment():
sys.exit(1)
# Check for required packages
print("\n📦 Checking required packages...")
try:
from mcp.server.fastmcp import FastMCP
print(" ✅ MCP (FastMCP) found")
except ImportError as e:
print(f" ❌ Missing MCP package: {e}")
print("\nInstall MCP with:")
print(" pip install mcp")
sys.exit(1)
try:
import httpx
print(" ✅ httpx found")
except ImportError:
print(" ❌ Missing httpx")
print("\nInstall with:")
print(" pip install httpx")
sys.exit(1)
try:
import pydantic
print(" ✅ pydantic found")
except ImportError as e:
print(f" ❌ Missing package: {e}")
print("\nInstall requirements with:")
print(" pip install -r requirements.txt")
sys.exit(1)
# Run async tests
print("\n🧪 Running functionality tests...")
success = asyncio.run(test_basic_functionality())
if not success:
sys.exit(1)
if __name__ == "__main__":
main()