| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179 |
- #!/usr/bin/env python3
- """
- Test script to reproduce the prompt validation error
- """
- import subprocess
- import json
- import sys
- import os
- def test_server_prompt():
- """Test the MCP server prompt functionality"""
- if len(sys.argv) != 2:
- print("Usage: python3 test_prompt_validation.py <server-executable>")
- sys.exit(1)
-
- server_path = sys.argv[1]
- if not os.path.exists(server_path):
- print(f"Error: Server executable '{server_path}' not found")
- sys.exit(1)
-
- print(f"Testing server: {server_path}")
-
- # Start the server process
- process = subprocess.Popen(
- [server_path],
- stdin=subprocess.PIPE,
- stdout=subprocess.PIPE,
- stderr=subprocess.PIPE,
- text=False
- )
-
- try:
- # Send initialize request
- init_request = {
- "jsonrpc": "2.0",
- "method": "initialize",
- "id": 0,
- "params": {
- "protocolVersion": "2025-11-25",
- "capabilities": {},
- "clientInfo": {
- "name": "test-client",
- "version": "1.0.0"
- }
- }
- }
-
- request_str = json.dumps(init_request)
- message = f"Content-Length: {len(request_str)}\r\n\r\n{request_str}"
- process.stdin.write(message.encode())
- process.stdin.flush()
-
- # Read response
- response_line = process.stdout.readline()
- if response_line.startswith(b"Content-Length:"):
- content_length = int(response_line.decode().split(":")[1].strip())
- process.stdout.readline() # Empty line
- response_data = process.stdout.read(content_length)
- response = json.loads(response_data.decode())
-
- if "result" in response:
- print("✓ Server initialized successfully")
- else:
- print("✗ Failed to initialize server")
- print(f"Response: {response}")
- return False
-
- # List prompts
- list_request = {
- "jsonrpc": "2.0",
- "method": "prompts/list",
- "id": 1,
- "params": {}
- }
-
- request_str = json.dumps(list_request)
- message = f"Content-Length: {len(request_str)}\r\n\r\n{request_str}"
- process.stdin.write(message.encode())
- process.stdin.flush()
-
- # Read response
- response_line = process.stdout.readline()
- if response_line.startswith(b"Content-Length:"):
- content_length = int(response_line.decode().split(":")[1].strip())
- process.stdout.readline() # Empty line
- response_data = process.stdout.read(content_length)
- response = json.loads(response_data.decode())
-
- if "result" in response and "prompts" in response["result"]:
- prompts = response["result"]["prompts"]
- print(f"✓ Found {len(prompts)} prompts")
-
- if prompts:
- # Get the first prompt
- prompt_name = prompts[0]["name"]
- print(f"Getting prompt: {prompt_name}")
-
- get_request = {
- "jsonrpc": "2.0",
- "method": "prompts/get",
- "id": 2,
- "params": {
- "name": prompt_name
- }
- }
-
- request_str = json.dumps(get_request)
- message = f"Content-Length: {len(request_str)}\r\n\r\n{request_str}"
- process.stdin.write(message.encode())
- process.stdin.flush()
-
- # Read response
- response_line = process.stdout.readline()
- if response_line.startswith(b"Content-Length:"):
- content_length = int(response_line.decode().split(":")[1].strip())
- process.stdout.readline() # Empty line
- response_data = process.stdout.read(content_length)
- response = json.loads(response_data.decode())
-
- if "result" in response:
- print("✓ Prompt retrieved successfully")
- result = response["result"]
-
- # Validate the structure
- if "messages" in result:
- messages = result["messages"]
- print(f"✓ Found {len(messages)} messages")
-
- for i, message in enumerate(messages):
- print(f"\nMessage {i}:")
- print(f" Role: {message.get('role')}")
-
- if "content" in message:
- content = message["content"]
- content_type = content.get("type")
- print(f" Content type: {content_type}")
- print(f" Content: {json.dumps(content, indent=2)}")
-
- # Check for validation issues
- if content_type == "text":
- if "text" not in content:
- print(" ✗ Missing 'text' field for text content")
- elif content_type in ["image", "audio"]:
- if "data" not in content:
- print(f" ✗ Missing 'data' field for {content_type} content")
- if "mimeType" not in content:
- print(f" ✗ Missing 'mimeType' field for {content_type} content")
- elif content_type == "resource_link":
- if "name" not in content:
- print(" ✗ Missing 'name' field for resource_link content")
- if "uri" not in content:
- print(" ✗ Missing 'uri' field for resource_link content")
- elif content_type == "resource":
- if "resource" not in content:
- print(" ✗ Missing 'resource' field for resource content")
- else:
- print("✗ No 'messages' field in response")
- else:
- print("✗ No 'result' field in response")
- if "error" in response:
- print(f"ERROR: {json.dumps(response['error'], indent=2)}")
- return False
- else:
- print("✗ No prompts found")
- else:
- print("✗ Failed to list prompts")
- print(f"Response: {response}")
- return False
-
- return True
-
- finally:
- process.terminate()
- process.wait()
- if __name__ == "__main__":
- success = test_server_prompt()
- sys.exit(0 if success else 1)
|