|
|
@@ -1,6 +1,7 @@
|
|
|
import hashlib
|
|
|
import json
|
|
|
import logging
|
|
|
+import socket
|
|
|
import threading
|
|
|
import time
|
|
|
import uuid
|
|
|
@@ -16,6 +17,45 @@ from utils.rate_limiter import SimpleRateLimiter
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
+UPLOAD_CORS_HEADERS = (
|
|
|
+ ('Access-Control-Allow-Origin', '*'),
|
|
|
+ ('Access-Control-Allow-Methods', 'POST, OPTIONS'),
|
|
|
+ ('Access-Control-Allow-Headers', 'Content-Type'),
|
|
|
+ ('Access-Control-Max-Age', '600'),
|
|
|
+)
|
|
|
+UPLOAD_BODY_LIMIT = 102 * 1024 * 1024
|
|
|
+UPLOAD_LIMIT_NAME = 'upload-headhaul-document'
|
|
|
+UPLOAD_DISCARD_CHUNK = 64 * 1024
|
|
|
+UPLOAD_DISCARD_IDLE_SECONDS = 0.5
|
|
|
+
|
|
|
+
|
|
|
+def discard_request_body(rfile, connection, length, chunk=UPLOAD_DISCARD_CHUNK,
|
|
|
+ idle_timeout=UPLOAD_DISCARD_IDLE_SECONDS):
|
|
|
+ if length <= 0 or rfile is None:
|
|
|
+ return
|
|
|
+ remaining = length
|
|
|
+ old_timeout = None
|
|
|
+ if connection is not None:
|
|
|
+ try:
|
|
|
+ old_timeout = connection.gettimeout()
|
|
|
+ connection.settimeout(idle_timeout)
|
|
|
+ except OSError:
|
|
|
+ old_timeout = None
|
|
|
+ try:
|
|
|
+ while remaining > 0:
|
|
|
+ data = rfile.read(min(chunk, remaining))
|
|
|
+ if not data:
|
|
|
+ break
|
|
|
+ remaining -= len(data)
|
|
|
+ except (OSError, socket.timeout):
|
|
|
+ pass
|
|
|
+ finally:
|
|
|
+ if connection is not None and old_timeout is not None:
|
|
|
+ try:
|
|
|
+ connection.settimeout(old_timeout)
|
|
|
+ except OSError:
|
|
|
+ pass
|
|
|
+
|
|
|
def extract_client_ip(headers, client_address):
|
|
|
if client_address and len(client_address) > 0:
|
|
|
return str(client_address[0])
|
|
|
@@ -101,6 +141,198 @@ class PublicMcpHttpHandler:
|
|
|
)
|
|
|
return None
|
|
|
|
|
|
+ @staticmethod
|
|
|
+ def _upload_error(code, message, trace_request_id):
|
|
|
+ return {
|
|
|
+ 'code': code,
|
|
|
+ 'msg': message,
|
|
|
+ 'data': {},
|
|
|
+ 'meta': {'request_id': trace_request_id},
|
|
|
+ }
|
|
|
+
|
|
|
+ def gate_upload(self, headers, content_length, client_ip='', trace_request_id=''):
|
|
|
+ if (
|
|
|
+ isinstance(content_length, bool)
|
|
|
+ or not isinstance(content_length, int)
|
|
|
+ or content_length < 0
|
|
|
+ ):
|
|
|
+ return 400, self._upload_error(
|
|
|
+ 'MCP_1401',
|
|
|
+ '工具参数不正确,请检查后重试',
|
|
|
+ trace_request_id,
|
|
|
+ ), None
|
|
|
+ if content_length > UPLOAD_BODY_LIMIT:
|
|
|
+ return 413, self._upload_error(
|
|
|
+ 'MCP_1401',
|
|
|
+ '文件不可大于100M',
|
|
|
+ trace_request_id,
|
|
|
+ ), None
|
|
|
+ context = self.context_parser.parse(headers or {})
|
|
|
+ session_id = context.gateway_session_id if context.has_session() else ''
|
|
|
+ identity = session_id or ('ip:' + str(client_ip or 'unknown'))
|
|
|
+ rate_key = '{0}:{1}'.format(identity, UPLOAD_LIMIT_NAME)
|
|
|
+ if self.rate_limiter is not None:
|
|
|
+ if not self.rate_limiter.is_allowed(rate_key):
|
|
|
+ logger.warning(
|
|
|
+ "MCP public rate limit exceeded",
|
|
|
+ extra={
|
|
|
+ 'request_id': trace_request_id,
|
|
|
+ 'protocol_method': 'upload',
|
|
|
+ 'tool_code': UPLOAD_LIMIT_NAME,
|
|
|
+ 'client_identity': identity,
|
|
|
+ 'diagnostic_reason': 'RATE_LIMIT_EXCEEDED',
|
|
|
+ },
|
|
|
+ )
|
|
|
+ return 429, self._upload_error(
|
|
|
+ 'MCP_9001',
|
|
|
+ '请求过于频繁,请稍后重试',
|
|
|
+ trace_request_id,
|
|
|
+ ), None
|
|
|
+ if not self.rate_limiter.try_acquire(rate_key):
|
|
|
+ logger.warning(
|
|
|
+ 'MCP public concurrency limit exceeded',
|
|
|
+ extra={
|
|
|
+ 'request_id': trace_request_id,
|
|
|
+ 'protocol_method': 'upload',
|
|
|
+ 'tool_code': UPLOAD_LIMIT_NAME,
|
|
|
+ 'client_identity': identity,
|
|
|
+ 'diagnostic_reason': 'CONCURRENCY_LIMIT_EXCEEDED',
|
|
|
+ },
|
|
|
+ )
|
|
|
+ return 429, self._upload_error(
|
|
|
+ 'MCP_9001',
|
|
|
+ '请求过于频繁,请稍后重试',
|
|
|
+ trace_request_id,
|
|
|
+ ), None
|
|
|
+ return None, None, rate_key
|
|
|
+
|
|
|
+ def handle_upload(
|
|
|
+ self,
|
|
|
+ headers,
|
|
|
+ body,
|
|
|
+ client_ip='',
|
|
|
+ trace_request_id='',
|
|
|
+ ):
|
|
|
+ if len(body or b'') > UPLOAD_BODY_LIMIT:
|
|
|
+ return 413, self._upload_error(
|
|
|
+ 'MCP_1401',
|
|
|
+ '文件不可大于100M',
|
|
|
+ trace_request_id,
|
|
|
+ )
|
|
|
+ fields, files = self._parse_multipart(headers, body or b'')
|
|
|
+ upload_token = str(fields.get('upload_token') or '').strip()
|
|
|
+ file_spec = files.get('file')
|
|
|
+ context = self.context_parser.parse(headers or {})
|
|
|
+ session_id = context.gateway_session_id if context.has_session() else ''
|
|
|
+ if (
|
|
|
+ not session_id
|
|
|
+ and upload_token
|
|
|
+ and hasattr(self.gateway_app, 'session_for_headhaul_token')
|
|
|
+ ):
|
|
|
+ session_id = str(
|
|
|
+ self.gateway_app.session_for_headhaul_token(upload_token) or ''
|
|
|
+ ).strip()
|
|
|
+ if not session_id:
|
|
|
+ return 401, {
|
|
|
+ 'code': 'MCP_1101',
|
|
|
+ 'msg': DEVICE_INVALID_MESSAGE,
|
|
|
+ 'data': {},
|
|
|
+ 'meta': {'request_id': trace_request_id},
|
|
|
+ }
|
|
|
+ if not upload_token or not file_spec:
|
|
|
+ return 400, {
|
|
|
+ 'code': 'MCP_1401',
|
|
|
+ 'msg': '工具参数不正确,请检查后重试',
|
|
|
+ 'data': {},
|
|
|
+ 'meta': {'request_id': trace_request_id},
|
|
|
+ }
|
|
|
+ filename, file_bytes, content_type = file_spec
|
|
|
+ try:
|
|
|
+ payload = self.gateway_app.upload_headhaul_document(
|
|
|
+ session_id,
|
|
|
+ upload_token,
|
|
|
+ filename,
|
|
|
+ file_bytes,
|
|
|
+ content_type,
|
|
|
+ request_id=trace_request_id,
|
|
|
+ client_ip=client_ip,
|
|
|
+ )
|
|
|
+ except RuntimeError as exc:
|
|
|
+ message = str(exc)
|
|
|
+ if message == DEVICE_INVALID_MESSAGE:
|
|
|
+ return 401, {
|
|
|
+ 'code': 'MCP_1101',
|
|
|
+ 'msg': DEVICE_INVALID_MESSAGE,
|
|
|
+ 'data': {},
|
|
|
+ 'meta': {'request_id': trace_request_id},
|
|
|
+ }
|
|
|
+ if message.lower().startswith('tool disabled'):
|
|
|
+ return 403, {
|
|
|
+ 'code': 'MCP_1202',
|
|
|
+ 'msg': '工具当前不可用',
|
|
|
+ 'data': {},
|
|
|
+ 'meta': {'request_id': trace_request_id},
|
|
|
+ }
|
|
|
+ return 500, {
|
|
|
+ 'code': 'MCP_9001',
|
|
|
+ 'msg': '系统繁忙,请稍后重试',
|
|
|
+ 'data': {},
|
|
|
+ 'meta': {'request_id': trace_request_id},
|
|
|
+ }
|
|
|
+ if not isinstance(payload, dict):
|
|
|
+ return 500, {
|
|
|
+ 'code': 'MCP_9001',
|
|
|
+ 'msg': '系统繁忙,请稍后重试',
|
|
|
+ 'data': {},
|
|
|
+ 'meta': {'request_id': trace_request_id},
|
|
|
+ }
|
|
|
+ return 200, payload
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def _parse_multipart(headers, body):
|
|
|
+ content_type = ''
|
|
|
+ for key, value in (headers or {}).items():
|
|
|
+ if str(key).lower() == 'content-type':
|
|
|
+ content_type = str(value or '')
|
|
|
+ break
|
|
|
+ if 'multipart/form-data' not in content_type or 'boundary=' not in content_type:
|
|
|
+ return {}, {}
|
|
|
+ boundary = content_type.split('boundary=', 1)[1].strip().strip('"')
|
|
|
+ delimiter = b'--' + boundary.encode('utf-8')
|
|
|
+ fields = {}
|
|
|
+ files = {}
|
|
|
+ for raw_part in body.split(delimiter):
|
|
|
+ part = raw_part.strip()
|
|
|
+ if not part or part == b'--':
|
|
|
+ continue
|
|
|
+ header_blob, separator, content = part.partition(b'\r\n\r\n')
|
|
|
+ if separator == b'':
|
|
|
+ continue
|
|
|
+ header_text = header_blob.decode('utf-8', 'replace')
|
|
|
+ disposition = ''
|
|
|
+ part_type = 'application/octet-stream'
|
|
|
+ for line in header_text.split('\r\n'):
|
|
|
+ lower = line.lower()
|
|
|
+ if lower.startswith('content-disposition:'):
|
|
|
+ disposition = line.split(':', 1)[1].strip()
|
|
|
+ if lower.startswith('content-type:'):
|
|
|
+ part_type = line.split(':', 1)[1].strip()
|
|
|
+ name = ''
|
|
|
+ filename = ''
|
|
|
+ for item in disposition.split(';'):
|
|
|
+ item = item.strip()
|
|
|
+ if item.startswith('name='):
|
|
|
+ name = item.split('=', 1)[1].strip().strip('"')
|
|
|
+ elif item.startswith('filename='):
|
|
|
+ filename = item.split('=', 1)[1].strip().strip('"')
|
|
|
+ if not name:
|
|
|
+ continue
|
|
|
+ if filename:
|
|
|
+ files[name] = (filename, content, part_type)
|
|
|
+ else:
|
|
|
+ fields[name] = content.decode('utf-8', 'replace')
|
|
|
+ return fields, files
|
|
|
+
|
|
|
def handle_json_rpc(
|
|
|
self,
|
|
|
headers,
|
|
|
@@ -145,7 +377,7 @@ class PublicMcpHttpHandler:
|
|
|
# rate-limiting it would block clients from connecting at all, so we skip it.
|
|
|
return McpProtocolHandler._success_response(request_id, {
|
|
|
'protocolVersion': McpProtocolHandler.protocol_version,
|
|
|
- 'capabilities': {'tools': {'listChanged': False}},
|
|
|
+ 'capabilities': McpProtocolHandler.server_capabilities(),
|
|
|
'serverInfo': {
|
|
|
'name': McpProtocolHandler.server_name,
|
|
|
'version': McpProtocolHandler.server_version,
|
|
|
@@ -197,6 +429,59 @@ class PublicMcpHttpHandler:
|
|
|
normalized['inputSchema'] = normalized.pop('input_schema')
|
|
|
tools.append(normalized)
|
|
|
return McpProtocolHandler._success_response(request_id, {'tools': tools})
|
|
|
+ if method in ('resources/list', 'resources/read'):
|
|
|
+ emitter.emit(
|
|
|
+ stage='protocol_validation',
|
|
|
+ status='succeeded',
|
|
|
+ event_code='PROTOCOL_VALIDATION_COMPLETED',
|
|
|
+ context={
|
|
|
+ 'jsonrpc_method': method,
|
|
|
+ 'transport': 'http',
|
|
|
+ },
|
|
|
+ )
|
|
|
+ protocol_validated = True
|
|
|
+ context = self.context_parser.parse(headers or {})
|
|
|
+ if not context.has_session():
|
|
|
+ emitter.emit(
|
|
|
+ stage='gateway_session',
|
|
|
+ status='failed',
|
|
|
+ event_code='GATEWAY_SESSION_NOT_FOUND',
|
|
|
+ context={'transport': 'http'},
|
|
|
+ )
|
|
|
+ return McpProtocolHandler._error_response(
|
|
|
+ request_id,
|
|
|
+ -32001,
|
|
|
+ DEVICE_INVALID_MESSAGE,
|
|
|
+ trace_request_id,
|
|
|
+ )
|
|
|
+ if method == 'resources/list':
|
|
|
+ resources = []
|
|
|
+ if hasattr(self.gateway_app, 'list_resources'):
|
|
|
+ resources = list(self.gateway_app.list_resources() or [])
|
|
|
+ return McpProtocolHandler._success_response(
|
|
|
+ request_id,
|
|
|
+ {'resources': resources},
|
|
|
+ )
|
|
|
+ uri = ''
|
|
|
+ if isinstance(message_params, dict):
|
|
|
+ uri = str(message_params.get('uri') or '').strip()
|
|
|
+ resource = None
|
|
|
+ if uri and hasattr(self.gateway_app, 'read_resource'):
|
|
|
+ resource = self.gateway_app.read_resource(
|
|
|
+ uri,
|
|
|
+ context.gateway_session_id,
|
|
|
+ )
|
|
|
+ if not isinstance(resource, dict):
|
|
|
+ return McpProtocolHandler._error_response(
|
|
|
+ request_id,
|
|
|
+ -32602,
|
|
|
+ 'Invalid params',
|
|
|
+ trace_request_id,
|
|
|
+ )
|
|
|
+ return McpProtocolHandler._success_response(
|
|
|
+ request_id,
|
|
|
+ {'contents': [resource]},
|
|
|
+ )
|
|
|
if method == 'tools/call':
|
|
|
if not isinstance(message_params, dict):
|
|
|
raise ValueError('tool parameters must be an object')
|
|
|
@@ -466,6 +751,16 @@ def create_http_handler(gateway_app, rate_limiter=None, reporter=None):
|
|
|
self.send_response(404)
|
|
|
self.end_headers()
|
|
|
|
|
|
+ def do_OPTIONS(self):
|
|
|
+ if self.path == '/mcp/upload-headhaul-document':
|
|
|
+ self.send_response(204)
|
|
|
+ for key, value in UPLOAD_CORS_HEADERS:
|
|
|
+ self.send_header(key, value)
|
|
|
+ self.end_headers()
|
|
|
+ return
|
|
|
+ self.send_response(404)
|
|
|
+ self.end_headers()
|
|
|
+
|
|
|
def do_POST(self):
|
|
|
request_headers = dict(self.headers.items())
|
|
|
client_ip = extract_client_ip(request_headers, self.client_address)
|
|
|
@@ -478,6 +773,35 @@ def create_http_handler(gateway_app, rate_limiter=None, reporter=None):
|
|
|
client_request_id_hash = rpc_handler._client_request_id(
|
|
|
request_headers
|
|
|
)
|
|
|
+ if self.path == '/mcp/upload-headhaul-document':
|
|
|
+ try:
|
|
|
+ length = int(self.headers.get('Content-Length') or '0')
|
|
|
+ except (TypeError, ValueError):
|
|
|
+ length = -1
|
|
|
+ status, payload, rate_key = rpc_handler.gate_upload(
|
|
|
+ request_headers,
|
|
|
+ length,
|
|
|
+ client_ip=client_ip,
|
|
|
+ trace_request_id=trace_request_id,
|
|
|
+ )
|
|
|
+ if status is not None:
|
|
|
+ if length > 0:
|
|
|
+ discard_request_body(self.rfile, self.connection, length)
|
|
|
+ self._write_upload_json(status, payload, close=True)
|
|
|
+ return
|
|
|
+ try:
|
|
|
+ body = self.rfile.read(length) if length > 0 else b''
|
|
|
+ status, payload = rpc_handler.handle_upload(
|
|
|
+ request_headers,
|
|
|
+ body,
|
|
|
+ client_ip=client_ip,
|
|
|
+ trace_request_id=trace_request_id,
|
|
|
+ )
|
|
|
+ self._write_upload_json(status, payload)
|
|
|
+ finally:
|
|
|
+ if rate_key and rpc_handler.rate_limiter is not None:
|
|
|
+ rpc_handler.rate_limiter.release(rate_key)
|
|
|
+ return
|
|
|
length = int(self.headers.get('Content-Length') or '0')
|
|
|
if self.path != '/mcp':
|
|
|
logger.warning(f"[HTTP] 404: ip={client_ip}, path={self.path}")
|
|
|
@@ -555,6 +879,19 @@ def create_http_handler(gateway_app, rate_limiter=None, reporter=None):
|
|
|
diagnostic_emitter=diagnostic_emitter,
|
|
|
)
|
|
|
|
|
|
+ def _write_upload_json(self, status, payload, close=False):
|
|
|
+ encoded = json.dumps(payload, ensure_ascii=False).encode('utf-8')
|
|
|
+ self.send_response(status)
|
|
|
+ self.send_header('Content-Type', 'application/json; charset=utf-8')
|
|
|
+ for key, value in UPLOAD_CORS_HEADERS:
|
|
|
+ self.send_header(key, value)
|
|
|
+ if close:
|
|
|
+ self.send_header('Connection', 'close')
|
|
|
+ self.close_connection = True
|
|
|
+ self.send_header('Content-Length', str(len(encoded)))
|
|
|
+ self.end_headers()
|
|
|
+ self.wfile.write(encoded)
|
|
|
+
|
|
|
def _write_json(self, payload, diagnostic_emitter=None):
|
|
|
raw = json.dumps(payload, ensure_ascii=False).encode('utf-8')
|
|
|
try:
|
|
|
@@ -613,7 +950,7 @@ def serve_public(gateway_app, host='0.0.0.0', port=8765, enable_rate_limit=True,
|
|
|
window_seconds=rate_limit_window_seconds,
|
|
|
max_in_flight=max_in_flight_per_tool,
|
|
|
)
|
|
|
- logger.info(f"Rate limiting enabled: {rate_limit_max_requests} requests/{rate_limit_window_seconds}s and {max_in_flight_per_tool} in-flight per session and tool (tools/call only)")
|
|
|
+ logger.info(f"Rate limiting enabled: {rate_limit_max_requests} requests/{rate_limit_window_seconds}s and {max_in_flight_per_tool} in-flight per session and tool (tools/call and upload HTTP)")
|
|
|
|
|
|
logger.info(f"Starting public MCP Gateway on {host}:{port}")
|
|
|
server = ThreadingHTTPServer(
|