test_prompt_validation.py 7.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179
  1. #!/usr/bin/env python3
  2. """
  3. Test script to reproduce the prompt validation error
  4. """
  5. import subprocess
  6. import json
  7. import sys
  8. import os
  9. def test_server_prompt():
  10. """Test the MCP server prompt functionality"""
  11. if len(sys.argv) != 2:
  12. print("Usage: python3 test_prompt_validation.py <server-executable>")
  13. sys.exit(1)
  14. server_path = sys.argv[1]
  15. if not os.path.exists(server_path):
  16. print(f"Error: Server executable '{server_path}' not found")
  17. sys.exit(1)
  18. print(f"Testing server: {server_path}")
  19. # Start the server process
  20. process = subprocess.Popen(
  21. [server_path],
  22. stdin=subprocess.PIPE,
  23. stdout=subprocess.PIPE,
  24. stderr=subprocess.PIPE,
  25. text=False
  26. )
  27. try:
  28. # Send initialize request
  29. init_request = {
  30. "jsonrpc": "2.0",
  31. "method": "initialize",
  32. "id": 0,
  33. "params": {
  34. "protocolVersion": "2025-11-25",
  35. "capabilities": {},
  36. "clientInfo": {
  37. "name": "test-client",
  38. "version": "1.0.0"
  39. }
  40. }
  41. }
  42. request_str = json.dumps(init_request)
  43. message = f"Content-Length: {len(request_str)}\r\n\r\n{request_str}"
  44. process.stdin.write(message.encode())
  45. process.stdin.flush()
  46. # Read response
  47. response_line = process.stdout.readline()
  48. if response_line.startswith(b"Content-Length:"):
  49. content_length = int(response_line.decode().split(":")[1].strip())
  50. process.stdout.readline() # Empty line
  51. response_data = process.stdout.read(content_length)
  52. response = json.loads(response_data.decode())
  53. if "result" in response:
  54. print("✓ Server initialized successfully")
  55. else:
  56. print("✗ Failed to initialize server")
  57. print(f"Response: {response}")
  58. return False
  59. # List prompts
  60. list_request = {
  61. "jsonrpc": "2.0",
  62. "method": "prompts/list",
  63. "id": 1,
  64. "params": {}
  65. }
  66. request_str = json.dumps(list_request)
  67. message = f"Content-Length: {len(request_str)}\r\n\r\n{request_str}"
  68. process.stdin.write(message.encode())
  69. process.stdin.flush()
  70. # Read response
  71. response_line = process.stdout.readline()
  72. if response_line.startswith(b"Content-Length:"):
  73. content_length = int(response_line.decode().split(":")[1].strip())
  74. process.stdout.readline() # Empty line
  75. response_data = process.stdout.read(content_length)
  76. response = json.loads(response_data.decode())
  77. if "result" in response and "prompts" in response["result"]:
  78. prompts = response["result"]["prompts"]
  79. print(f"✓ Found {len(prompts)} prompts")
  80. if prompts:
  81. # Get the first prompt
  82. prompt_name = prompts[0]["name"]
  83. print(f"Getting prompt: {prompt_name}")
  84. get_request = {
  85. "jsonrpc": "2.0",
  86. "method": "prompts/get",
  87. "id": 2,
  88. "params": {
  89. "name": prompt_name
  90. }
  91. }
  92. request_str = json.dumps(get_request)
  93. message = f"Content-Length: {len(request_str)}\r\n\r\n{request_str}"
  94. process.stdin.write(message.encode())
  95. process.stdin.flush()
  96. # Read response
  97. response_line = process.stdout.readline()
  98. if response_line.startswith(b"Content-Length:"):
  99. content_length = int(response_line.decode().split(":")[1].strip())
  100. process.stdout.readline() # Empty line
  101. response_data = process.stdout.read(content_length)
  102. response = json.loads(response_data.decode())
  103. if "result" in response:
  104. print("✓ Prompt retrieved successfully")
  105. result = response["result"]
  106. # Validate the structure
  107. if "messages" in result:
  108. messages = result["messages"]
  109. print(f"✓ Found {len(messages)} messages")
  110. for i, message in enumerate(messages):
  111. print(f"\nMessage {i}:")
  112. print(f" Role: {message.get('role')}")
  113. if "content" in message:
  114. content = message["content"]
  115. content_type = content.get("type")
  116. print(f" Content type: {content_type}")
  117. print(f" Content: {json.dumps(content, indent=2)}")
  118. # Check for validation issues
  119. if content_type == "text":
  120. if "text" not in content:
  121. print(" ✗ Missing 'text' field for text content")
  122. elif content_type in ["image", "audio"]:
  123. if "data" not in content:
  124. print(f" ✗ Missing 'data' field for {content_type} content")
  125. if "mimeType" not in content:
  126. print(f" ✗ Missing 'mimeType' field for {content_type} content")
  127. elif content_type == "resource_link":
  128. if "name" not in content:
  129. print(" ✗ Missing 'name' field for resource_link content")
  130. if "uri" not in content:
  131. print(" ✗ Missing 'uri' field for resource_link content")
  132. elif content_type == "resource":
  133. if "resource" not in content:
  134. print(" ✗ Missing 'resource' field for resource content")
  135. else:
  136. print("✗ No 'messages' field in response")
  137. else:
  138. print("✗ No 'result' field in response")
  139. if "error" in response:
  140. print(f"ERROR: {json.dumps(response['error'], indent=2)}")
  141. return False
  142. else:
  143. print("✗ No prompts found")
  144. else:
  145. print("✗ Failed to list prompts")
  146. print(f"Response: {response}")
  147. return False
  148. return True
  149. finally:
  150. process.terminate()
  151. process.wait()
  152. if __name__ == "__main__":
  153. success = test_server_prompt()
  154. sys.exit(0 if success else 1)