#!/usr/bin/env python3
"""Client for connecting to MCP server running as systemd service.
For systemd services, the recommended approach is to use the HTTP server
since stdio communication is difficult with background services.
This script provides utilities to:
1. Check service status
2. Connect via HTTP (if HTTP server is also running)
3. Use systemd journal for communication (if configured)
"""
import subprocess
import sys
import json
import requests
from typing import Dict, Any, Optional
def check_service_status() -> Dict[str, Any]:
"""Check if the systemd service is running."""
try:
result = subprocess.run(
["systemctl", "is-active", "42crunch-mcp.service"],
capture_output=True,
text=True,
timeout=5
)
is_active = result.returncode == 0 and result.stdout.strip() == "active"
# Get service status
status_result = subprocess.run(
["systemctl", "status", "42crunch-mcp.service", "--no-pager"],
capture_output=True,
text=True,
timeout=5
)
return {
"active": is_active,
"status_output": status_result.stdout,
"service_name": "42crunch-mcp.service"
}
except FileNotFoundError:
return {
"active": False,
"error": "systemctl not found (not on Linux or systemd not available)",
"service_name": "42crunch-mcp.service"
}
except Exception as e:
return {
"active": False,
"error": str(e),
"service_name": "42crunch-mcp.service"
}
def connect_via_http(base_url: str = "http://localhost:8000") -> Optional[requests.Session]:
"""Connect to HTTP server (recommended for systemd services)."""
try:
response = requests.get(f"{base_url}/health", timeout=5)
if response.status_code == 200:
print(f"✅ HTTP server is running at {base_url}")
return requests.Session()
except requests.exceptions.ConnectionError:
print(f"❌ HTTP server not available at {base_url}")
print(" Start HTTP server with: python http_main.py")
except Exception as e:
print(f"❌ Error connecting to HTTP server: {e}")
return None
def send_request_via_http(session: requests.Session, base_url: str, method: str, params: Dict[str, Any] = None) -> Dict[str, Any]:
"""Send JSON-RPC request via HTTP."""
request = {
"jsonrpc": "2.0",
"id": 1,
"method": method,
}
if params:
request["params"] = params
response = session.post(
f"{base_url}/jsonrpc",
json=request,
headers={"Content-Type": "application/json"},
timeout=30
)
response.raise_for_status()
return response.json()
def main():
"""Main function to test connection to systemd service."""
import argparse
parser = argparse.ArgumentParser(
description="Connect to MCP server running as systemd service",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
# Check service status
python test_client_systemd.py --status
# Connect via HTTP (recommended)
python test_client_systemd.py --http
# Test with HTTP server
python test_client_systemd.py --http --test
"""
)
parser.add_argument(
"--status",
action="store_true",
help="Check systemd service status"
)
parser.add_argument(
"--http",
action="store_true",
help="Connect via HTTP server (recommended for systemd services)"
)
parser.add_argument(
"--http-url",
type=str,
default="http://localhost:8000",
help="HTTP server URL (default: http://localhost:8000)"
)
parser.add_argument(
"--test",
action="store_true",
help="Run test requests"
)
args = parser.parse_args()
print("="*60)
print("Systemd Service MCP Client")
print("="*60)
# Check service status
if args.status or not args.http:
print("\n📋 Checking systemd service status...")
status = check_service_status()
if status.get("active"):
print(f"✅ Service '{status['service_name']}' is ACTIVE")
else:
print(f"❌ Service '{status['service_name']}' is NOT ACTIVE")
if "error" in status:
print(f" Error: {status['error']}")
print("\n💡 To start the service:")
print(" sudo systemctl start 42crunch-mcp.service")
print(" sudo systemctl enable 42crunch-mcp.service")
# Connect via HTTP (recommended for systemd services)
if args.http:
print("\n🌐 Connecting via HTTP server...")
print(" (HTTP is recommended for systemd services)")
session = connect_via_http(args.http_url)
if session and args.test:
print("\n🧪 Running tests...")
# Test list_collections
try:
response = send_request_via_http(
session,
args.http_url,
"tools/call",
{
"name": "list_collections",
"arguments": {"per_page": 5}
}
)
print(f"✅ list_collections: {response.get('result', {}).get('success', False)}")
except Exception as e:
print(f"❌ list_collections failed: {e}")
elif session:
print("✅ Connected! Use --test to run tests")
else:
print("\n💡 To use HTTP server:")
print(" 1. Start HTTP server: python http_main.py")
print(" 2. Or configure systemd service to run HTTP server")
print(" 3. Then use: python test_client_systemd.py --http --test")
# If no options provided, show help
if not args.status and not args.http:
print("\n💡 Usage options:")
print(" --status : Check systemd service status")
print(" --http : Connect via HTTP server (recommended)")
print(" --http --test : Connect and run tests")
print("\n⚠️ Note: Direct stdio connection to systemd services is not supported.")
print(" Use HTTP server for systemd services.")
if __name__ == "__main__":
main()
