"""
HTTP based Remote MCP Server entry point
Compliant with MCP specification 2025-03-26
- Streamable HTTP transport
- Stateless server (no session management)
- Remote server (publicly accessible URL)
"""
import os
import json
import logging
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.middleware.cors import CORSMiddleware
from mcp.server import Server
from contextlib import asynccontextmanager
from src.tools import register_tools
from src.tools.link_scanner import (
scan_video_link,
scan_text_link,
get_scan_video_link_tool,
get_scan_text_link_tool
)
from src.resources import register_resources
from mcp.types import TextContent
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Create MCP server instance
mcp_server = Server("link-scan")
# Register tools and resources
register_tools(mcp_server)
register_resources(mcp_server)
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup
logger.info("Starting Link Scan MCP Server (Remote)...")
yield
# Shutdown
logger.info("Shutting down Link Scan MCP Server...")
app = FastAPI(
title="Link Scan MCP Server",
description="Remote MCP Server for link scanning and summarization",
version="1.0.0",
lifespan=lifespan
)
# Add CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# API prefix for multi-server hosting
API_PREFIX = os.getenv("API_PREFIX", "/link-scan")
@app.get("/")
async def root():
"""Root endpoint with server information"""
return {
"name": "link-scan",
"version": "1.0.0",
"status": "running",
"description": "Remote MCP Server for link scanning and summarization",
"mcp_version": "2025-03-26",
"api_prefix": API_PREFIX,
"endpoint": f"{API_PREFIX}/messages"
}
@app.get(f"{API_PREFIX}/health")
async def health():
"""Health check endpoint"""
return {"status": "healthy", "service": "link-scan"}
@app.post(f"{API_PREFIX}/messages")
async def messages_endpoint(request: Request):
"""
MCP protocol messages endpoint
Handles JSON-RPC style messages from MCP clients
"""
try:
body = await request.json()
# Extract method and params from JSON-RPC request
method = body.get("method")
params = body.get("params", {})
request_id = body.get("id")
# Handle different MCP methods
if method == "initialize":
# Return server capabilities
response = {
"jsonrpc": "2.0",
"id": request_id,
"result": {
"protocolVersion": "2025-03-26",
"capabilities": {
"tools": {}
},
"serverInfo": {
"name": "link-scan",
"version": "1.0.0"
}
}
}
return JSONResponse(content=response)
elif method == "tools/list":
# Get list of tools
video_tool = get_scan_video_link_tool()
text_tool = get_scan_text_link_tool()
tools_list = [
{
"name": video_tool.name,
"description": video_tool.description,
"inputSchema": video_tool.inputSchema
},
{
"name": text_tool.name,
"description": text_tool.description,
"inputSchema": text_tool.inputSchema
}
]
response = {
"jsonrpc": "2.0",
"id": request_id,
"result": {
"tools": tools_list
}
}
return JSONResponse(content=response)
elif method == "tools/call":
# Call a tool
tool_name = params.get("name")
tool_args = params.get("arguments", {})
url = tool_args.get("url", "")
if not url:
response = {
"jsonrpc": "2.0",
"id": request_id,
"error": {
"code": -32602,
"message": "url parameter is required"
}
}
return JSONResponse(content=response, status_code=400)
try:
if tool_name == "scan_video_link":
result = await scan_video_link(url)
elif tool_name == "scan_text_link":
result = await scan_text_link(url)
else:
response = {
"jsonrpc": "2.0",
"id": request_id,
"error": {
"code": -32601,
"message": f"Tool not found: {tool_name}"
}
}
return JSONResponse(content=response, status_code=404)
# Enforce 24k character limit (PlayMCP policy requirement)
MAX_RESPONSE_SIZE = 24 * 1024 # 24k characters
if len(result) > MAX_RESPONSE_SIZE:
logger.warning(f"Response exceeds 24k limit ({len(result)} chars), truncating to {MAX_RESPONSE_SIZE} chars")
result = result[:MAX_RESPONSE_SIZE - 3] + "..."
content = [TextContent(type="text", text=result)]
response = {
"jsonrpc": "2.0",
"id": request_id,
"result": {
"content": [
{
"type": item.type,
"text": item.text
}
for item in content
]
}
}
return JSONResponse(content=response)
except Exception as e:
# If scan_video_link fails, try fallback to scan_text_link
if tool_name == "scan_video_link":
try:
logger.warning(f"Video link scan failed, falling back to text link scan: {str(e)}")
result = await scan_text_link(url)
# Enforce 24k character limit (PlayMCP policy requirement)
MAX_RESPONSE_SIZE = 24 * 1024 # 24k characters
if len(result) > MAX_RESPONSE_SIZE:
logger.warning(f"Response exceeds 24k limit ({len(result)} chars), truncating to {MAX_RESPONSE_SIZE} chars")
result = result[:MAX_RESPONSE_SIZE - 3] + "..."
content = [TextContent(type="text", text=result)]
response = {
"jsonrpc": "2.0",
"id": request_id,
"result": {
"content": [
{
"type": item.type,
"text": item.text
}
for item in content
]
}
}
return JSONResponse(content=response)
except Exception as fallback_error:
logger.error(f"Error calling tool {tool_name} and fallback failed: {str(e)}, {str(fallback_error)}")
response = {
"jsonrpc": "2.0",
"id": request_id,
"error": {
"code": -32603,
"message": f"Internal error: Video processing failed ({str(e)}), and text fallback also failed ({str(fallback_error)})"
}
}
return JSONResponse(content=response, status_code=500)
logger.error(f"Error calling tool {tool_name}: {str(e)}")
response = {
"jsonrpc": "2.0",
"id": request_id,
"error": {
"code": -32603,
"message": f"Internal error: {str(e)}"
}
}
return JSONResponse(content=response, status_code=500)
else:
# Unknown method
response = {
"jsonrpc": "2.0",
"id": request_id,
"error": {
"code": -32601,
"message": f"Method not found: {method}"
}
}
return JSONResponse(content=response, status_code=404)
except json.JSONDecodeError:
return JSONResponse(
content={"error": "Invalid JSON"},
status_code=400
)
except Exception as e:
logger.error(f"Error in messages endpoint: {str(e)}")
return JSONResponse(
content={"error": str(e)},
status_code=500
)
if __name__ == "__main__":
import uvicorn
port = int(os.getenv("PORT", 8000))
host = os.getenv("HOST", "0.0.0.0")
uvicorn.run(
"src.server_http:app",
host=host,
port=port,
reload=os.getenv("DEBUG", "False").lower() == "true"
)
