main.py•3.07 kB
#!/usr/bin/env python3
"""
Main entry point for MCP Data Fetch Server.
"""
import asyncio
import json
import sys
import logging
import argparse
from pathlib import Path
from server import MCPDataFetchServer
from models import MCPMessage
from config import DEFAULT_WORKING_DIR, DEFAULT_CACHE_SUBDIR
# Logging Configuration
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s",
handlers=[logging.StreamHandler(sys.stderr)]
)
logger = logging.getLogger(__name__)
def parse_args() -> argparse.Namespace:
"""Parse command line arguments."""
parser = argparse.ArgumentParser(
description="MCP Data Fetch Server - Secure web content fetching with sandboxed working directory"
)
parser.add_argument(
"-d", "--working-dir", type=Path, default=DEFAULT_WORKING_DIR,
help=f"Sandboxed working directory (default: {DEFAULT_WORKING_DIR})"
)
parser.add_argument(
"-c", "--cache-dir", type=str, default=DEFAULT_CACHE_SUBDIR,
help=f"Cache subdirectory name relative to working directory (default: {DEFAULT_CACHE_SUBDIR})"
)
return parser.parse_args()
async def main() -> None:
"""Main loop."""
args = parse_args()
# Setup sandboxed working directory
working_dir = args.working_dir.resolve()
working_dir.mkdir(parents=True, exist_ok=True)
# Setup cache directory (relative to working directory)
cache_dir = working_dir / args.cache_dir
cache_dir.mkdir(parents=True, exist_ok=True)
logger.info(f"Working directory: {working_dir}")
logger.info(f"Cache directory: {cache_dir}")
server = MCPDataFetchServer(working_dir=working_dir, cache_dir=cache_dir)
try:
while True:
try:
line = await asyncio.get_event_loop().run_in_executor(
None, sys.stdin.readline
)
if not line:
break
line = line.strip()
if not line:
continue
try:
message_dict = json.loads(line)
message = MCPMessage.from_dict(message_dict)
except json.JSONDecodeError as exc:
logger.error(f"Invalid JSON: {exc}")
continue
response = await server.handle_message(message)
if response:
print(json.dumps(response.to_dict()), flush=True)
except (EOFError, KeyboardInterrupt):
break
except Exception as exc:
logger.error(f"Unexpected error: {exc}", exc_info=True)
finally:
await server.close()
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
logger.info("Server terminated")
except Exception as exc:
logger.error(f"Fatal error: {exc}", exc_info=True)
sys.exit(1)