mcp-cortellis

Verified
MIT License
Overview InspectNew Schema Related Servers Reviews Score
mcp-cortellis
examples
"""
Test script for MCP methods against local Docker container.
Tests list_tools, initialize, search_drugs, and explore_ontology methods.

Authentication:
    This script requires the CORTELLIS_AUTH environment variable to be set with a base64 encoded
    token in the format base64(username:password). You can generate this token using:
    
    # Unix/macOS:
    echo -n "your_username:your_password" | base64
    
    # Windows PowerShell:
    [Convert]::ToBase64String([Text.Encoding]::UTF8.GetBytes("your_username:your_password"))
"""
import os
import asyncio
import json
import logging
import websockets

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

async def test_method(websocket, method, params=None):
    """Test a specific WebSocket method."""
    request = {
        "jsonrpc": "2.0",
        "method": method,
        "id": 1
    }
    if params:
        request["params"] = params
    
    logger.info(f"\nTesting method: {method}")
    logger.info(f"Sending request: {json.dumps(request, indent=2)}")
    
    await websocket.send(json.dumps(request))
    response = await websocket.recv()
    response_data = json.loads(response)
    
    logger.info(f"Received response: {json.dumps(response_data, indent=2)}")
    return response_data

async def send_notification(websocket, method, params=None):
    """Send a notification (no response expected)."""
    request = {
        "jsonrpc": "2.0",
        "method": method
    }
    if params:
        request["params"] = params
    
    logger.info(f"\nSending notification: {method}")
    logger.info(f"Request: {json.dumps(request, indent=2)}")
    
    await websocket.send(json.dumps(request))

async def run_tests():
    """Run all MCP method tests."""
    # Verify authentication token is set
    if not os.getenv('CORTELLIS_AUTH'):
        logger.error("CORTELLIS_AUTH environment variable not set")
        logger.error("Please set it with your base64 encoded token (base64(username:password))")
        return

    uri = "ws://localhost:8765"
    logger.info(f"Connecting to {uri}")
    
    try:
        async with websockets.connect(uri) as websocket:
            # Test initialize with MCP capabilities
            init_params = {
                "protocol_version": "1.0.0",
                "capabilities": {
                    "transport": ["ws"],
                    "methods": ["search_drugs", "explore_ontology"]
                }
            }
            init_response = await test_method(websocket, "initialize", init_params)
            if "error" in init_response:
                logger.error("Initialization failed - check your authentication token")
                return
            
            # Send initialized notification
            await send_notification(websocket, "initialized")
            
            # Test search_drugs
            search_params = {
                "query": "obesity",
                "phase": "C3"
            }
            await test_method(websocket, "search_drugs", search_params)
            
            # Test explore_ontology
            ontology_params = {
                "term": "GLP-1",
                "category": "action"
            }
            await test_method(websocket, "explore_ontology", ontology_params)
            
            # Send shutdown notification
            await send_notification(websocket, "shutdown")
            
    except websockets.exceptions.ConnectionClosed as e:
        logger.error(f"WebSocket connection closed: {e}")
    except Exception as e:
        logger.error(f"Error during testing: {e}")

if __name__ == "__main__":
    asyncio.run(run_tests()) 