mcp-cortellis

Verified
MIT License
Overview InspectNew Schema Related Servers Reviews Score
mcp-cortellis
examples
"""
Script to test the capabilities endpoint of the local WebSocket server.

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 websockets
import logging

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

async def test_server():
    """Test the server endpoints."""
    # 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 None

    uri = "ws://localhost:8765"
    logger.info(f"Connecting to {uri}")
    
    try:
        async with websockets.connect(uri) as websocket:
            # Test initialize first
            initialize_request = {
                "jsonrpc": "2.0",
                "method": "initialize",
                "id": 1
            }
            
            logger.info(f"Sending initialize request: {initialize_request}")
            await websocket.send(json.dumps(initialize_request))
            
            logger.info("Waiting for initialize response...")
            response = await websocket.recv()
            response_data = json.loads(response)
            logger.info(f"Received response: {json.dumps(response_data, indent=2)}")
            
            if "error" in response_data:
                logger.error("Initialization failed")
                return None

            # Test list_tools
            list_tools_request = {
                "jsonrpc": "2.0",
                "method": "list_tools",
                "id": 2
            }
            
            logger.info(f"Sending list_tools request: {list_tools_request}")
            await websocket.send(json.dumps(list_tools_request))
            
            logger.info("Waiting for list_tools response...")
            response = await websocket.recv()
            response_data = json.loads(response)
            logger.info(f"Received response: {json.dumps(response_data, indent=2)}")
            
            # Test inspect
            inspect_request = {
                "jsonrpc": "2.0",
                "method": "inspect",
                "params": {
                    "tool": "cortellis_mcp"
                },
                "id": 3
            }
            
            logger.info(f"Sending inspect request: {inspect_request}")
            await websocket.send(json.dumps(inspect_request))
            
            logger.info("Waiting for inspect response...")
            response = await websocket.recv()
            response_data = json.loads(response)
            logger.info(f"Received response: {json.dumps(response_data, indent=2)}")
            
            return response_data
    except Exception as e:
        logger.error(f"Error: {e}")
        return None

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