test_tools.py•2.43 kB
#!/usr/bin/env python3
"""Test script to call YouTube Search MCP server tools"""
import subprocess
import json
import sys
import os
def test_youtube_search():
"""Test the YouTube search tool"""
current_dir = os.path.dirname(os.path.abspath(__file__))
venv_python = os.path.join(current_dir, 'venv', 'bin', 'python')
server_script = os.path.join(current_dir, 'youtube_search_mcp.py')
# Initialize message
init_message = {
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {"tools": {}},
"clientInfo": {"name": "test-client", "version": "1.0.0"}
}
}
# Get tools list
tools_message = {
"jsonrpc": "2.0",
"id": 2,
"method": "tools/list",
"params": {}
}
# Test search (you need to set YOUTUBE_API_KEY)
search_message = {
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {
"name": "search_youtube",
"arguments": {
"query": "python tutorial",
"max_results": 3
}
}
}
try:
process = subprocess.Popen(
[venv_python, server_script],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
env={**os.environ, "YOUTUBE_API_KEY": "your_api_key_here"}
)
# Send all messages
messages = [init_message, tools_message, search_message]
input_data = '\n'.join(json.dumps(msg) for msg in messages) + '\n'
stdout, stderr = process.communicate(input=input_data, timeout=15)
print("=== Server Response ===")
for line in stdout.strip().split('\n'):
if line.strip():
try:
response = json.loads(line)
print(json.dumps(response, indent=2))
print("---")
except json.JSONDecodeError:
print(f"Raw line: {line}")
if stderr:
print("\n=== Server Errors ===")
print(stderr)
except Exception as e:
print(f"Error: {e}")
if __name__ == "__main__":
print("Testing YouTube Search Tools...")
test_youtube_search()