"""A small MCP server over stdio, for trying out and testing the MCP console. Run it through the bridge (see mcp.config.example.json), or by hand: python3 scripts/mcp_demo_server.py and type JSON-RPC messages, one per line. Everything the protocol needs is here in plain Python (no SDK), so it also serves as a readable reference for writing a server. """ import base64 import json import sys import threading import time import zlib PROTOCOL_VERSIONS = ['2025-06-18', '2025-03-26', '2024-11-05'] PAGE_SIZE = 3 # tools/list is paginated so clients have to follow nextCursor TOOLS = [ {'name': 'echo', 'title': '回声', 'description': 'Repeat a text a number of times.', 'inputSchema': {'type': 'object', 'properties': { 'text': {'type': 'string', 'description': '要重复的文字'}, 'times': {'type': 'integer', 'minimum': 1, 'maximum': 10, 'default': 1}}, 'required': ['text']}}, {'name': 'add', 'title': '加法', 'description': 'Add two numbers.', 'inputSchema': {'type': 'object', 'properties': {'a': {'type': 'number'}, 'b': {'type': 'number'}}, 'required': ['a', 'b']}}, {'name': 'image', 'title': '生成图片', 'description': 'Return a small PNG gradient in the given colour.', 'inputSchema': {'type': 'object', 'properties': {'color': {'type': 'string', 'enum': ['red', 'green', 'blue']}}, 'required': ['color']}}, {'name': 'fail', 'title': '报错', 'description': 'Always report a tool error (isError).', 'inputSchema': {'type': 'object', 'properties': {}}}, {'name': 'slow', 'title': '慢调用', 'description': 'Sleep before answering (to try timeouts).', 'inputSchema': {'type': 'object', 'properties': {'seconds': {'type': 'number', 'minimum': 0, 'maximum': 60}}, 'required': ['seconds']}}, {'name': 'notify', 'title': '推送通知', 'description': 'Send log and progress notifications while working.', 'inputSchema': {'type': 'object', 'properties': {'steps': {'type': 'integer', 'minimum': 1, 'maximum': 10, 'default': 3}}}}, {'name': 'ask_client', 'title': '反向请求', 'description': 'Ask the client for its roots (a server-to-client request) and report the answer.', 'inputSchema': {'type': 'object', 'properties': {}}}, {'name': 'crash', 'title': '崩溃', 'description': 'Exit the server process (to try restarts).', 'inputSchema': {'type': 'object', 'properties': {}}}, ] RESOURCES = [{'uri': 'demo://readme', 'name': 'readme', 'title': '说明', 'mimeType': 'text/markdown'}] TEMPLATES = [{'uriTemplate': 'demo://notes/{name}', 'name': 'note', 'title': '笔记', 'mimeType': 'text/plain'}] PROMPTS = [{'name': 'review', 'title': '代码审查', 'description': 'Ask for a review of a code snippet.', 'arguments': [{'name': 'code', 'description': '要审查的代码', 'required': True}, {'name': 'language', 'description': '语言', 'required': False}]}] write_lock = threading.Lock() pending = {} # id of a request this server sent → [threading.Event, response] next_request_id = [1000] def send(message): with write_lock: sys.stdout.write(json.dumps(message, ensure_ascii=False) + '\n') sys.stdout.flush() def png(color): """A 32×32 RGB gradient PNG, built by hand.""" channel = {'red': 0, 'green': 1, 'blue': 2}[color] rows = b'' for y in range(32): row = bytearray([0]) for x in range(32): px = [40, 40, 40] px[channel] = 60 + x * 6 row += bytes(px) rows += bytes(row) chunk = lambda kind, data: len(data).to_bytes(4, 'big') + kind + data + zlib.crc32(kind + data).to_bytes(4, 'big') ihdr = (32).to_bytes(4, 'big') * 2 + bytes([8, 2, 0, 0, 0]) return b'\x89PNG\r\n\x1a\n' + chunk(b'IHDR', ihdr) + chunk(b'IDAT', zlib.compress(rows)) + chunk(b'IEND', b'') def text(t, is_error=False): return {'content': [{'type': 'text', 'text': t}], 'isError': is_error} def ask_client(method, params, timeout=10): """Send a request to the client and wait for its response.""" next_request_id[0] += 1 rid = next_request_id[0] waiter = [threading.Event(), None] pending[rid] = waiter send({'jsonrpc': '2.0', 'id': rid, 'method': method, 'params': params}) if not waiter[0].wait(timeout): pending.pop(rid, None) return None return waiter[1] def call_tool(name, args, progress_token): if name == 'echo': return text(' '.join([args['text']] * int(args.get('times', 1)))) if name == 'add': return text(str(args['a'] + args['b'])) if name == 'image': return {'content': [{'type': 'image', 'data': base64.b64encode(png(args['color'])).decode(), 'mimeType': 'image/png'}, {'type': 'text', 'text': f"32×32 {args['color']} gradient"}]} if name == 'fail': return text('This tool always fails (on purpose).', is_error=True) if name == 'slow': time.sleep(float(args['seconds'])) return text(f"slept {args['seconds']} s") if name == 'notify': steps = int(args.get('steps', 3)) for k in range(1, steps + 1): send({'jsonrpc': '2.0', 'method': 'notifications/message', 'params': {'level': 'info', 'logger': 'demo', 'data': f'step {k}/{steps}'}}) if progress_token is not None: send({'jsonrpc': '2.0', 'method': 'notifications/progress', 'params': {'progressToken': progress_token, 'progress': k, 'total': steps}}) time.sleep(0.05) return text(f'done after {steps} steps') if name == 'ask_client': reply = ask_client('roots/list', {}) if reply is None: return text('the client did not answer within 10 s', is_error=True) return text('client answered: ' + json.dumps(reply, ensure_ascii=False)) if name == 'crash': print('crashing on purpose', file=sys.stderr, flush=True) sys.stdout.flush() import os os._exit(3) raise KeyError(name) def handle(msg): method, mid, params = msg.get('method'), msg.get('id'), msg.get('params') or {} if method is None: # a response to one of our requests waiter = pending.pop(mid, None) if waiter: waiter[1] = msg waiter[0].set() return if mid is None: # notification: nothing to answer return def reply(result=None, error=None): send({'jsonrpc': '2.0', 'id': mid, **({'error': error} if error else {'result': result})}) try: if method == 'initialize': asked = params.get('protocolVersion') reply({'protocolVersion': asked if asked in PROTOCOL_VERSIONS else PROTOCOL_VERSIONS[0], 'capabilities': {'tools': {'listChanged': False}, 'resources': {}, 'prompts': {}, 'logging': {}}, 'serverInfo': {'name': 'webtt-demo', 'title': '示例服务端', 'version': '1.0.0'}, 'instructions': 'A demo server: try echo, image, notify and ask_client.'}) elif method == 'ping': reply({}) elif method == 'tools/list': start = int(params.get('cursor') or 0) page = {'tools': TOOLS[start:start + PAGE_SIZE]} if start + PAGE_SIZE < len(TOOLS): page['nextCursor'] = str(start + PAGE_SIZE) reply(page) elif method == 'tools/call': name = params.get('name') if name not in {t['name'] for t in TOOLS}: reply(error={'code': -32602, 'message': f'Unknown tool: {name}'}) return token = (params.get('_meta') or {}).get('progressToken') # Slow tools run on their own thread so the server keeps reading (e.g. the # client's answer to ask_client) def run(): try: reply(call_tool(name, params.get('arguments') or {}, token)) except (KeyError, TypeError, ValueError) as e: reply(error={'code': -32602, 'message': f'Invalid arguments: {e}'}) threading.Thread(target=run, daemon=True).start() elif method == 'resources/list': reply({'resources': RESOURCES}) elif method == 'resources/templates/list': reply({'resourceTemplates': TEMPLATES}) elif method == 'resources/read': uri = params.get('uri', '') if uri == 'demo://readme': reply({'contents': [{'uri': uri, 'mimeType': 'text/markdown', 'text': '# 示例服务端\n\n用来试用 MCP 调试台。'}]}) elif uri.startswith('demo://notes/'): reply({'contents': [{'uri': uri, 'mimeType': 'text/plain', 'text': f'笔记：{uri[13:]}'}]}) else: reply(error={'code': -32002, 'message': f'Resource not found: {uri}'}) elif method == 'prompts/list': reply({'prompts': PROMPTS}) elif method == 'prompts/get': args = params.get('arguments') or {} if 'code' not in args: reply(error={'code': -32602, 'message': 'missing argument: code'}) return lang = args.get('language', '') reply({'description': 'Code review', 'messages': [{'role': 'user', 'content': {'type': 'text', 'text': f'请审查这段{lang}代码：\n{args["code"]}'}}]}) elif method == 'logging/setLevel': reply({}) else: reply(error={'code': -32601, 'message': f'Method not found: {method}'}) except Exception as e: # a bug here should show up in the console, not kill the server reply(error={'code': -32603, 'message': f'Internal error: {e}'}) def main(): print('demo MCP server ready', file=sys.stderr, flush=True) for line in sys.stdin: line = line.strip() if not line: continue try: msg = json.loads(line) except json.JSONDecodeError as e: send({'jsonrpc': '2.0', 'id': None, 'error': {'code': -32700, 'message': f'Parse error: {e}'}}) continue handle(msg) if __name__ == '__main__': main()