diagnose.py•8.18 kB
#!/usr/bin/env python3
"""
Enhanced Diagnostic script for Drip MCP Server
This script helps identify connection and configuration issues.
"""
import os
import sys
import asyncio
import json
import base64
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"))
print("="*60)
print("Drip MCP Server - Enhanced Diagnostic Tool")
print("="*60)
# 1. Check environment variables
print("\n1. Environment Variables Check:")
api_key = os.environ.get("DRIP_API_KEY", "")
account_id = os.environ.get("DRIP_ACCOUNT_ID", "")
if api_key and account_id:
print(f" ✅ DRIP_API_KEY: {'*' * 10}{api_key[-4:]}")
print(f" ✅ DRIP_ACCOUNT_ID: {account_id}")
else:
print(" ❌ Missing environment variables")
if not api_key:
print(" Missing: DRIP_API_KEY")
if not account_id:
print(" Missing: DRIP_ACCOUNT_ID")
sys.exit(1)
# 2. Test direct API connection with multiple approaches
print("\n2. Testing Direct API Connection:")
async def test_connection():
import httpx
# Build auth header
auth_string = f"{api_key}:"
encoded_auth = base64.b64encode(auth_string.encode()).decode()
headers = {
"Authorization": f"Basic {encoded_auth}",
"Content-Type": "application/json",
"User-Agent": "Drip MCP Diagnostic Tool",
"Accept": "application/json"
}
print(f"\n Auth Token (last 4): ...{encoded_auth[-4:]}")
# Test 1: Try the accounts list endpoint (doesn't need account ID)
print("\n Test 1: Listing all accounts...")
url1 = "https://api.getdrip.com/v2/accounts"
try:
async with httpx.AsyncClient(verify=False, timeout=30.0, follow_redirects=True) as client:
print(f" URL: {url1}")
response = await client.get(url1, headers=headers)
print(f" Status: {response.status_code}")
if response.status_code == 200:
data = response.json()
if "accounts" in data:
print(f" ✅ Successfully authenticated with Drip API")
print(f" Found {len(data['accounts'])} account(s):")
for acc in data["accounts"]:
acc_id = acc.get('id', 'Unknown')
acc_name = acc.get('name', 'Unknown')
print(f" - ID: {acc_id}, Name: {acc_name}")
if str(acc_id) == str(account_id):
print(f" ✅ Your account found!")
return True
if len(data['accounts']) > 0:
print(f"\n ⚠️ Your account ID ({account_id}) was not found in the list")
print(f" Available account IDs: {', '.join(str(acc.get('id')) for acc in data['accounts'])}")
return False
return True
elif response.status_code == 401:
print(" ❌ Authentication failed")
print(" Your API key appears to be invalid")
print(" Please check:")
print(" 1. You're using the API Token, not your password")
print(" 2. The token hasn't been regenerated/revoked")
print(" 3. Copy the entire token without spaces")
if response.text:
print(f" API Response: {response.text[:200]}")
return False
else:
print(f" ❌ Unexpected response: {response.status_code}")
if response.text:
print(f" Response: {response.text[:300]}")
return False
except httpx.ConnectError as e:
print(f" ❌ Connection failed: {str(e)[:200]}")
print("\n Possible causes:")
print(" 1. Network/firewall blocking HTTPS requests")
print(" 2. Proxy settings interfering")
print(" 3. DNS resolution issues")
# Try alternative test
print("\n Test 2: Testing basic HTTPS connectivity...")
try:
async with httpx.AsyncClient(verify=False, timeout=10.0) as client:
test_response = await client.get("https://httpbin.org/get")
if test_response.status_code == 200:
print(" ✅ General HTTPS works")
print(" ❌ But cannot reach Drip API specifically")
else:
print(" ❌ General HTTPS connectivity issue")
except:
print(" ❌ Cannot make any HTTPS requests")
print(" Check your internet connection and firewall")
return False
except httpx.TimeoutException:
print(" ❌ Request timed out after 30 seconds")
print(" The Drip API might be temporarily unavailable")
return False
except Exception as e:
print(f" ❌ Unexpected error: {type(e).__name__}: {str(e)[:200]}")
return False
# Run the connection test
api_works = asyncio.run(test_connection())
if not api_works:
print("\n" + "="*60)
print("❌ Cannot connect to Drip API")
print("\n📋 Troubleshooting Checklist:")
print("\n1. Verify API Token:")
print(" - Go to Drip: Settings → My User Settings")
print(" - Find 'API Token' section")
print(" - Click 'Regenerate' if needed")
print(" - Copy the ENTIRE token")
print("\n2. Verify Account ID:")
print(" - Look at your Drip dashboard URL")
print(" - Example: https://app.drip.com/6600640/...")
print(" - The number (6600640) is your account ID")
print("\n3. Test with curl:")
print(f" curl -H 'Authorization: Basic {base64.b64encode(f'{api_key}:'.encode()).decode()}' \\")
print(" https://api.getdrip.com/v2/accounts")
print("\n4. Check Drip Status:")
print(" - Visit: https://status.drip.com/")
print(" - Check for any API outages")
print("="*60)
sys.exit(1)
# 3. Test the MCP server functions
print("\n3. Testing MCP Server Functions:")
try:
# Test imports
from drip_mcp import make_api_request, get_auth_headers, API_BASE_URL
print(" ✅ Server module imported")
# Show configuration
print(f"\n Configuration:")
print(f" API Base URL: {API_BASE_URL}")
# Test auth headers
try:
headers = get_auth_headers()
print(f" ✅ Auth headers generated")
except Exception as e:
print(f" ❌ Auth header error: {e}")
# Test a simple API call
async def test_api_call():
try:
print("\n Testing make_api_request...")
result = await make_api_request("GET", "{account_id}/tags")
print(f" ✅ API request successful")
if isinstance(result, dict):
if 'tags' in result:
print(f" Tags found: {len(result['tags'])}")
if result['tags']:
print(f" Sample tags: {', '.join(result['tags'][:5])}")
except Exception as e:
print(f" ❌ API request failed: {e}")
asyncio.run(test_api_call())
except ImportError as e:
print(f" ❌ Failed to import server module: {e}")
print(" Ensure you're running from the project root directory")
print(" Run: python scripts/diagnose.py")
except Exception as e:
print(f" ❌ Unexpected error: {e}")
print("\n" + "="*60)
if api_works:
print("✅ Diagnostic Complete - API Connection Successful!")
print("\nYour Drip MCP server should work correctly.")
print("\nTo use with Claude Code:")
print("1. Configure Claude Code with your credentials")
print("2. Start the server: python -m drip_mcp.server")
print("3. Use natural language commands in Claude Code")
else:
print("❌ Diagnostic Complete - Connection Issues Found")
print("\nPlease resolve the issues above before using the server.")
print("="*60)