292 lines
9.7 KiB
Python
292 lines
9.7 KiB
Python
import requests
|
||
import json
|
||
from pathlib import Path
|
||
import yaml
|
||
# ─────────────────────────────
|
||
# 🗂️ Credential + Config Paths
|
||
# ─────────────────────────────
|
||
|
||
|
||
def load_mcp_config():
|
||
"""Load MCP configuration from file located in the same directory as this script."""
|
||
# Always use the current script’s directory
|
||
script_dir = Path(__file__).resolve().parent
|
||
mcp_config_path = script_dir / "mcp_config.json"
|
||
|
||
if not mcp_config_path.exists():
|
||
print(f"⚠️ MCP config not found at {mcp_config_path}")
|
||
return None
|
||
|
||
try:
|
||
with open(mcp_config_path, "r", encoding="utf-8") as f:
|
||
return json.load(f)
|
||
except (json.JSONDecodeError, OSError) as e:
|
||
print(f"⚠️ Failed to load MCP config: {e}")
|
||
return None
|
||
|
||
|
||
def get_mcp_tools_metadata():
|
||
"""
|
||
Get metadata for all available MCP tools.
|
||
|
||
Returns:
|
||
List of tool metadata dictionaries, or None if failed
|
||
"""
|
||
mcp_config = load_mcp_config()
|
||
if not mcp_config:
|
||
print("⚠️ MCP config unavailable")
|
||
return None
|
||
|
||
try:
|
||
server_url = mcp_config["url"].rstrip("/")
|
||
access_token = mcp_config["token"]
|
||
|
||
# MCP protocol endpoint
|
||
mcp_endpoint = f"{server_url}/mcp/"
|
||
|
||
# Construct MCP JSON-RPC call to list tools
|
||
payload = {
|
||
"jsonrpc": "2.0",
|
||
"id": 1,
|
||
"method": "tools/list",
|
||
"params": {}
|
||
}
|
||
|
||
# Make the request with SSE support
|
||
response = requests.post(
|
||
mcp_endpoint,
|
||
headers={
|
||
"Authorization": f"Bearer {access_token}",
|
||
"Content-Type": "application/json",
|
||
"Accept": "application/json, text/event-stream"
|
||
},
|
||
json=payload,
|
||
timeout=10,
|
||
stream=True
|
||
)
|
||
|
||
response.raise_for_status()
|
||
|
||
# Handle SSE stream response
|
||
if response.headers.get('content-type', '').startswith('text/event-stream'):
|
||
# Parse SSE stream
|
||
for line in response.iter_lines():
|
||
if line:
|
||
decoded = line.decode('utf-8')
|
||
if decoded.startswith('data: '):
|
||
data = decoded[6:] # Remove 'data: ' prefix
|
||
try:
|
||
event_data = json.loads(data)
|
||
|
||
# Check for JSON-RPC error
|
||
if "error" in event_data:
|
||
print(f"⚠️ MCP error: {event_data['error']}")
|
||
return None
|
||
|
||
# Return the tools list when found
|
||
if event_data.get("jsonrpc") == "2.0" and "result" in event_data:
|
||
result = event_data["result"]
|
||
# The result contains a "tools" array
|
||
if "tools" in result:
|
||
return result["tools"]
|
||
return result
|
||
except json.JSONDecodeError:
|
||
continue
|
||
return None
|
||
else:
|
||
# Handle regular JSON response (fallback)
|
||
result = response.json()
|
||
|
||
# Check for JSON-RPC error
|
||
if "error" in result:
|
||
print(f"⚠️ MCP error: {result['error']}")
|
||
return None
|
||
|
||
# Return the tools list
|
||
if "result" in result:
|
||
if "tools" in result["result"]:
|
||
return result["result"]["tools"]
|
||
return result["result"]
|
||
|
||
return result
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
print(f"⚠️ MCP call failed: {e}")
|
||
return None
|
||
except Exception as e:
|
||
print(f"⚠️ Unexpected error: {e}")
|
||
return None
|
||
|
||
|
||
def print_tools_metadata(tools):
|
||
"""Pretty print tools metadata."""
|
||
if not tools:
|
||
print("No tools found")
|
||
return
|
||
|
||
print(f"\n📋 Found {len(tools)} tool(s):\n")
|
||
for i, tool in enumerate(tools, 1):
|
||
print(f"{'='*60}")
|
||
print(f"Tool #{i}: {tool.get('name', 'Unknown')}")
|
||
print(f"{'='*60}")
|
||
print(f"Description: {tool.get('description', 'No description')}")
|
||
|
||
# Print input schema
|
||
if 'inputSchema' in tool or 'input_schema' in tool:
|
||
schema = tool.get('inputSchema') or tool.get('input_schema')
|
||
print(f"\n📝 Input Schema:")
|
||
print(f" Type: {schema.get('type', 'N/A')}")
|
||
|
||
if 'properties' in schema:
|
||
print(f" Properties:")
|
||
for prop_name, prop_info in schema['properties'].items():
|
||
prop_type = prop_info.get('type', 'unknown')
|
||
prop_title = prop_info.get('title', prop_name)
|
||
print(f" - {prop_name} ({prop_type}): {prop_title}")
|
||
|
||
if 'required' in schema:
|
||
print(f" Required: {', '.join(schema['required'])}")
|
||
|
||
# Print metadata (your custom meta field)
|
||
if 'meta' in tool:
|
||
print(f"\n🏷️ Metadata:")
|
||
for key, value in tool['meta'].items():
|
||
print(f" {key}: {value}")
|
||
|
||
# Print annotations
|
||
if 'annotations' in tool:
|
||
print(f"\n🔖 Annotations:")
|
||
for key, value in tool['annotations'].items():
|
||
print(f" {key}: {value}")
|
||
|
||
print()
|
||
|
||
|
||
|
||
def get_manual_mcp_response(tool_name: str, arguments: dict):
|
||
"""
|
||
Manually call an MCP tool using the MCP protocol.
|
||
|
||
Args:
|
||
tool_name: Name of the tool to call (e.g., "play_sound_effect")
|
||
arguments: Dictionary of arguments for the tool (e.g., {"sound_type": "bong"})
|
||
|
||
Returns:
|
||
The tool result or None if failed
|
||
"""
|
||
mcp_config = load_mcp_config()
|
||
if not mcp_config:
|
||
print("⚠️ MCP config unavailable")
|
||
return None
|
||
|
||
try:
|
||
server_url = mcp_config["url"].rstrip("/")
|
||
access_token = mcp_config["token"]
|
||
|
||
# MCP protocol endpoint
|
||
mcp_endpoint = f"{server_url}/mcp/"
|
||
|
||
# Construct MCP JSON-RPC call
|
||
payload = {
|
||
"jsonrpc": "2.0",
|
||
"id": 1,
|
||
"method": "tools/call",
|
||
"params": {
|
||
"name": tool_name,
|
||
"arguments": arguments
|
||
}
|
||
}
|
||
|
||
# Make the request with SSE support
|
||
response = requests.post(
|
||
mcp_endpoint,
|
||
headers={
|
||
"Authorization": f"Bearer {access_token}",
|
||
"Content-Type": "application/json",
|
||
"Accept": "application/json, text/event-stream"
|
||
},
|
||
json=payload,
|
||
timeout=10,
|
||
stream=True # Important for SSE
|
||
)
|
||
|
||
response.raise_for_status()
|
||
|
||
# Handle SSE stream response
|
||
if response.headers.get('content-type', '').startswith('text/event-stream'):
|
||
# Parse SSE stream
|
||
for line in response.iter_lines():
|
||
if line:
|
||
decoded = line.decode('utf-8')
|
||
if decoded.startswith('data: '):
|
||
data = decoded[6:] # Remove 'data: ' prefix
|
||
try:
|
||
event_data = json.loads(data)
|
||
|
||
# Check for JSON-RPC error
|
||
if "error" in event_data:
|
||
print(f"⚠️ MCP error: {event_data['error']}")
|
||
return None
|
||
|
||
# Return the result when found
|
||
if event_data.get("jsonrpc") == "2.0" and "result" in event_data:
|
||
return event_data["result"]
|
||
except json.JSONDecodeError:
|
||
continue
|
||
return None
|
||
else:
|
||
# Handle regular JSON response (fallback)
|
||
result = response.json()
|
||
|
||
# Check for JSON-RPC error
|
||
if "error" in result:
|
||
print(f"⚠️ MCP error: {result['error']}")
|
||
return None
|
||
|
||
# Return the actual result
|
||
if "result" in result:
|
||
return result["result"]
|
||
|
||
return result
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
print(f"⚠️ MCP call failed: {e}")
|
||
return None
|
||
except Exception as e:
|
||
print(f"⚠️ Unexpected error: {e}")
|
||
return None
|
||
|
||
|
||
|
||
|
||
# Example usage
|
||
if __name__ == "__main__":
|
||
print("🔍 Fetching MCP tools metadata...")
|
||
tools = get_mcp_tools_metadata()
|
||
|
||
if tools:
|
||
print_tools_metadata(tools)
|
||
|
||
# Also print raw JSON for inspection
|
||
print("\n" + "="*60)
|
||
print("📦 Raw JSON Response:")
|
||
print("="*60)
|
||
print(json.dumps(tools, indent=2))
|
||
else:
|
||
print("❌ Failed to fetch tools metadata")
|
||
|
||
|
||
print("Starte Google Kalender Test...")
|
||
result = get_manual_mcp_response(
|
||
tool_name="get_calendar_events",
|
||
arguments={"max_results": 5},
|
||
)
|
||
print("Kalender Resultat:", result)
|
||
|
||
# manual call argument
|
||
result = get_manual_mcp_response(
|
||
tool_name="roll_dice",
|
||
arguments={"n_dice": 3, "manual_call": True},
|
||
)
|
||
print("Manually called Result:", result)
|