kris/afv-library
1
0
Fork 0
mirror of https://github.com/forcedotcom/afv-library.git synced 2026-08-11 18:38:22 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

Delete skills/check_mcp.py

Browse Source
This commit is contained in:
thrylokya 2026-03-20 09:05:22 +05:30 committed by GitHub
parent 1db330b2f3
commit 59d7edde88
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 0 additions and 444 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

444
skills/check_mcp.py
View File

@ -1,444 +0,0 @@
#!/usr/bin/env python3
"""
Check MCP Content Server Availability and Tools
Returns JSON with availability status for each media source:
{
"cms_search": true|false,
"data_cloud": true|false,
"unsplash": true|false
}
Invocation: Run with `python3 check_mcp.py` or, if Python is not available,
use the Bash equivalent `bash check_mcp.sh`. Both produce identical output.
"""
import json
import os
import sys
from pathlib import Path
from urllib.request import Request, urlopen
from urllib.error import HTTPError, URLError
# Use only stdlib (no requests) so the script runs in minimal Vibes/CI environments
# Timeout for HTTP requests (seconds)
TIMEOUT = 5
# Relative path from IDE config root (Code/Cursor) to A4D MCP settings file
_A4D_MCP_SETTINGS_REL = Path('User/globalStorage/salesforce.salesforcedx-einstein-gpt/settings/a4d_mcp_settings.json')
def _get_ide_config_root():
"""
Return the IDE (Code/VS Code) config root directory for this platform.
Used to find globalStorage for the Salesforce Einstein GPT extension.
"""
if sys.platform == 'win32':
appdata = os.environ.get('APPDATA', '')
if not appdata:
appdata = Path.home() / 'AppData' / 'Roaming'
return Path(appdata) / 'Code'
if sys.platform == 'darwin':
return Path.home() / 'Library' / 'Application Support' / 'Code'
# Linux and other POSIX
xdg = os.environ.get('XDG_CONFIG_HOME') or (Path.home() / '.config')
return Path(xdg) / 'Code'
def get_a4d_mcp_settings_path():
"""Return the A4D MCP settings path for the current platform (Mac, Windows, Linux, etc.)."""
return _get_ide_config_root() / _A4D_MCP_SETTINGS_REL
# Path to A4D MCP settings (platform-agnostic)
SETTINGS_PATH = get_a4d_mcp_settings_path()
# URL patterns that identify the content MCP endpoint (server name is irrelevant)
CONTENT_ENDPOINT_PATTERN = 'platform/content'
CONTENT_READONLY_PATTERN = 'content-readonly'
# Args pattern that identifies the Unsplash MCP server (server name is irrelevant)
UNSPLASH_MCP_ARGS_PATTERN = 'unsplash-mcp-server'
def _http_post_json(url, headers, payload, timeout=TIMEOUT):
"""
POST JSON to url using stdlib only (no requests dependency).
Returns (status_code, response_headers, response_data_dict or None).
On failure returns (None, None, None).
"""
data = json.dumps(payload).encode('utf-8')
req_headers = {**headers, 'Content-Type': 'application/json'}
req = Request(url, data=data, headers=req_headers, method='POST')
try:
with urlopen(req, timeout=timeout) as resp:
body = resp.read().decode('utf-8')
result = json.loads(body) if body else None
return (resp.status, resp.headers, result)
except HTTPError as e:
try:
body = e.read().decode('utf-8')
result = json.loads(body) if body else None
except Exception:
result = None
return (e.code, e.headers, result)
except (URLError, OSError, ValueError, json.JSONDecodeError):
return (None, None, None)
def get_raw_settings():
"""
Read and parse the full A4D MCP settings file.
Returns the full settings dict or None if not found / invalid.
"""
try:
if not SETTINGS_PATH.exists():
return None
with open(SETTINGS_PATH, 'r') as f:
return json.load(f)
except Exception:
return None
def get_mcp_settings():
"""
Read and parse A4D MCP settings.
Returns the mcpServers dict or None if not found.
"""
settings = get_raw_settings()
if not settings:
return None
return settings.get('mcpServers', {})
def _args_contain_content_endpoint(args):
"""True if args list contains the content MCP endpoint URL (any server name)."""
if not args:
return False
for arg in args:
if isinstance(arg, str) and (
CONTENT_ENDPOINT_PATTERN in arg or CONTENT_READONLY_PATTERN in arg
):
return True
return False
def get_content_server_config():
"""
Get the MCP server config that uses the content endpoint URL.
Does not depend on server name; finds any server whose args contain
the content endpoint (e.g. .../platform/content or .../content-readonly).
Returns the config dict or empty dict if not found.
"""
mcp_servers = get_mcp_settings()
if not mcp_servers:
return {}
for config in mcp_servers.values():
if not isinstance(config, dict):
continue
args = config.get('args', [])
if _args_contain_content_endpoint(args):
return config
return {}
def get_contentmcp_url():
"""
Extract MCP content server URL from settings.
Returns the URL or None if not found, not configured, or stdio type.
"""
content_config = get_content_server_config()
if not content_config:
return None
if content_config.get('type') == 'stdio':
return None
args = content_config.get('args', [])
for arg in args:
if isinstance(arg, str) and (
CONTENT_ENDPOINT_PATTERN in arg or CONTENT_READONLY_PATTERN in arg
):
return arg
return None
def is_contentmcp_enabled():
"""
Check if content MCP server is enabled in settings.
Returns True if enabled (disabled: false) AND not stdio type, False otherwise.
Note: For stdio type servers, we cannot check availability via URL,
so we return False to skip URL-based checks.
"""
content_config = get_content_server_config()
if not content_config:
return False
# Check if disabled field is explicitly false (enabled)
disabled = content_config.get('disabled', False)
if disabled:
return False
# If it's stdio type, we can't do URL-based availability checks
# The MCP server is managed by the IDE/extension, not accessible via HTTP
if content_config.get('type') == 'stdio':
# For stdio, assume it's available if enabled
# (The IDE manages the connection)
return True
# For non-stdio (like mcp-remote), we can check via URL
return True
def _args_contain_unsplash_mcp(args):
"""True if args list contains the unsplash MCP server (any server name)."""
if not args:
return False
for arg in args:
if isinstance(arg, str) and UNSPLASH_MCP_ARGS_PATTERN in arg:
return True
return False
def get_unsplash_server_config():
"""
Get the Unsplash MCP server config from settings.
Checks (1) root-level "unsplash" key and (2) any server in mcpServers
whose args contain "unsplash-mcp-server" (name-agnostic).
Returns the config dict or empty dict if not found.
"""
settings = get_raw_settings()
if not settings:
return {}
# Root-level "unsplash" (some configs put it here)
root_unsplash = settings.get('unsplash')
if isinstance(root_unsplash, dict) and root_unsplash:
return root_unsplash
# Any server in mcpServers that runs unsplash-mcp-server
mcp_servers = settings.get('mcpServers', {})
for config in mcp_servers.values() if mcp_servers else ():
if isinstance(config, dict) and _args_contain_unsplash_mcp(config.get('args', [])):
return config
return {}
def is_unsplash_mcp_enabled():
"""
Check if Unsplash MCP server is configured and enabled.
Returns True if configured and enabled, False otherwise.
"""
unsplash_config = get_unsplash_server_config()
if not unsplash_config:
return False
disabled = unsplash_config.get('disabled', False)
return not disabled
def get_bearer_token():
"""
Extract bearer token from A4D MCP settings.
Returns the token string or None if not found.
"""
content_config = get_content_server_config()
if not content_config:
return None
# Extract Authorization header from args
args = content_config.get('args', [])
for i, arg in enumerate(args):
if arg == '--header' and i + 1 < len(args):
auth_header = args[i + 1]
# Format: "Authorization: Bearer <token>"
if auth_header.startswith('Authorization: Bearer '):
return auth_header.replace('Authorization: Bearer ', '')
return None
def get_base_headers():
"""Get base headers for MCP requests (without session ID)."""
headers = {
'Content-Type': 'application/json',
'Accept': 'application/json, text/event-stream'
}
token = get_bearer_token()
if token:
headers['Authorization'] = f'Bearer {token}'
return headers
def initialize_mcp_session(endpoint_url):
"""
Initialize MCP session and get session ID.
Args:
endpoint_url: The MCP server endpoint URL
Returns session ID or None if initialization fails.
"""
headers = get_base_headers()
# Step 1: Initialize
payload = {
"jsonrpc": "2.0",
"method": "initialize",
"id": "0",
"params": {
"protocolVersion": "2025-03-26",
"capabilities": {},
"clientInfo": {"name": "CheckMCPScript", "version": "1.0.0"}
}
}
try:
status, resp_headers, data = _http_post_json(endpoint_url, headers, payload, TIMEOUT)
if status == 200 and data is not None:
# Try to get from Mcp-Session-Id header
session_id = resp_headers.get('Mcp-Session-Id') if resp_headers else None
if session_id:
return session_id
# Try to get from response body
if 'result' in data and isinstance(data['result'], dict):
session_id = data['result'].get('sessionId') or data['result'].get('session_id')
if session_id:
return session_id
except Exception:
pass
return None
def send_initialized_notification(endpoint_url, session_id):
"""
Send initialized notification to MCP server.
Args:
endpoint_url: The MCP server endpoint URL
session_id: The session ID from initialization
Returns True if successful, False otherwise.
"""
headers = get_base_headers()
headers['Mcp-Session-Id'] = session_id
payload = {
"jsonrpc": "2.0",
"method": "notifications/initialized"
}
try:
status, _, _ = _http_post_json(endpoint_url, headers, payload, TIMEOUT)
return status in [200, 202] # Accept both 200 and 202
except Exception:
return False
def list_available_tools():
"""
List all available tools from MCP server using JSON-RPC.
Follows proper MCP initialization flow.
Returns a list of tool names, or empty list if unavailable.
"""
# Get the endpoint URL
endpoint_url = get_contentmcp_url()
if not endpoint_url:
# stdio type or no URL configured
return []
# Step 1: Initialize and get session ID
session_id = initialize_mcp_session(endpoint_url)
if not session_id:
return []
# Step 2: Send initialized notification
if not send_initialized_notification(endpoint_url, session_id):
return []
# Step 3: List tools
headers = get_base_headers()
headers['Mcp-Session-Id'] = session_id
payload = {
"jsonrpc": "2.0",
"method": "tools/list",
"id": "1",
"params": {}
}
try:
status, _, data = _http_post_json(endpoint_url, headers, payload, TIMEOUT)
if status == 200 and data and 'result' in data and 'tools' in data['result']:
tools = data['result']['tools']
tool_names = [tool.get('name') for tool in tools if isinstance(tool, dict) and 'name' in tool]
return tool_names
except Exception:
pass
return []
def check_mcp_health():
"""Check if MCP server is reachable by attempting to list tools."""
tools = list_available_tools()
return len(tools) > 0
def check_tool_available(tool_name):
"""
Check if a specific tool is available via MCP server.
Uses JSON-RPC to list tools and searches for the tool name.
"""
available_tools = list_available_tools()
return tool_name in available_tools
def main():
"""Main execution."""
result = {
"cms_search": False,
"data_cloud": False,
"unsplash": False
}
# First check if contentmcp server is enabled in settings
if not is_contentmcp_enabled():
# contentmcp is disabled, return all false
print(json.dumps(result))
return 0
# Check if it's stdio type
content_config = get_content_server_config()
if content_config.get('type') == 'stdio':
# For stdio type, we can't check via URL
# Assume tools are available if the server is enabled
# The IDE/extension manages the connection
result["cms_search"] = True
result["data_cloud"] = True
else:
# For non-stdio (like mcp-remote), check via URL
if check_mcp_health():
# Check for CMS search tool
if check_tool_available("search_media_cms_channels"):
result["cms_search"] = True
# Check for Data Cloud search tool
if check_tool_available("search_electronic_media"):
result["data_cloud"] = True
# Check for Unsplash MCP server separately
if is_unsplash_mcp_enabled():
result["unsplash"] = True
# Output JSON result
print(json.dumps(result))
return 0
if __name__ == "__main__":
sys.exit(main())