test_wrapper.py•4.7 kB
#!/usr/bin/env python3
"""
Test script for Crawl4AI MCP wrapper.
Verifies that all tools can communicate with the Crawl4AI Docker container.
"""
import asyncio
import sys
import httpx
CRAWL4AI_BASE_URL = "http://localhost:11235"
async def test_health():
"""Test that Crawl4AI Docker container is running."""
print("Testing Crawl4AI container health...")
async with httpx.AsyncClient() as client:
try:
response = await client.get(f"{CRAWL4AI_BASE_URL}/health")
response.raise_for_status()
data = response.json()
print(f"✅ Container healthy: {data}")
return True
except Exception as e:
print(f"❌ Container not responding: {e}")
return False
async def test_scrape():
"""Test markdown scraping."""
print("\nTesting scrape_markdown tool...")
async with httpx.AsyncClient(timeout=60.0) as client:
try:
response = await client.post(
f"{CRAWL4AI_BASE_URL}/md",
json={"url": "https://example.com", "f": "fit"}
)
response.raise_for_status()
data = response.json()
if data.get("success") and "markdown" in data:
preview = data["markdown"][:100]
print(f"✅ Scrape successful. Preview: {preview}...")
return True
else:
print(f"❌ Scrape failed: {data}")
return False
except Exception as e:
print(f"❌ Scrape error: {e}")
return False
async def test_html():
"""Test HTML extraction."""
print("\nTesting extract_html tool...")
async with httpx.AsyncClient(timeout=60.0) as client:
try:
response = await client.post(
f"{CRAWL4AI_BASE_URL}/html",
json={"url": "https://example.com"}
)
response.raise_for_status()
data = response.json()
print(f"✅ HTML extraction successful")
return True
except Exception as e:
print(f"❌ HTML extraction error: {e}")
return False
async def test_screenshot():
"""Test screenshot capture."""
print("\nTesting capture_screenshot tool...")
async with httpx.AsyncClient(timeout=90.0) as client:
try:
response = await client.post(
f"{CRAWL4AI_BASE_URL}/screenshot",
json={"url": "https://example.com", "screenshot_wait_for": 2}
)
response.raise_for_status()
data = response.json()
print(f"✅ Screenshot captured successfully")
return True
except Exception as e:
print(f"❌ Screenshot error: {e}")
return False
async def test_crawl():
"""Test multi-URL crawling."""
print("\nTesting crawl_urls tool...")
async with httpx.AsyncClient(timeout=120.0) as client:
try:
response = await client.post(
f"{CRAWL4AI_BASE_URL}/crawl",
json={"urls": ["https://example.com"]}
)
response.raise_for_status()
data = response.json()
print(f"✅ Crawl successful")
return True
except Exception as e:
print(f"❌ Crawl error: {e}")
return False
async def main():
"""Run all tests."""
print("=" * 60)
print("Crawl4AI MCP Wrapper Test Suite")
print("=" * 60)
# Test container health first
if not await test_health():
print("\n❌ Crawl4AI container is not running!")
print("Start it with: docker start crawl4ai")
sys.exit(1)
# Run all tool tests
results = []
results.append(("Scrape Markdown", await test_scrape()))
results.append(("Extract HTML", await test_html()))
results.append(("Capture Screenshot", await test_screenshot()))
results.append(("Crawl URLs", await test_crawl()))
# Summary
print("\n" + "=" * 60)
print("Test Summary")
print("=" * 60)
for test_name, passed in results:
status = "✅ PASS" if passed else "❌ FAIL"
print(f"{status}: {test_name}")
all_passed = all(result for _, result in results)
print("\n" + "=" * 60)
if all_passed:
print("✅ All tests passed! MCP wrapper is ready to use.")
print("\nNext steps:")
print("1. Configure Claude Code with the MCP server")
print("2. Restart Claude Code")
print("3. Test using: claude mcp list")
else:
print("❌ Some tests failed. Check the output above.")
sys.exit(1)
if __name__ == "__main__":
asyncio.run(main())