Brunobkr/llama.cpp_AlgMor24_github

1#!/usr/bin/env python32"""3MCP server that crashes after receiving a specific tool call.4"""5import json6import sys7import os8 9def handle_initialize(params, req_id):10    return {11        "jsonrpc": "2.0",12        "id": req_id,13        "result": {14            "protocolVersion": "2024-11-05",15            "capabilities": {"tools": {}},16            "serverInfo": {"name": "crash-test", "version": "1.0"}17        }18    }19 20def handle_tools_list(params, req_id):21    return {22        "jsonrpc": "2.0",23        "id": req_id,24        "result": {25            "tools": [26                {27                    "name": "echo",28                    "description": "Echo back the input message",29                    "inputSchema": {30                        "type": "object",31                        "properties": {32                            "message": {"type": "string"}33                        }34                    }35                },36                {37                    "name": "crash",38                    "description": "Crash the server",39                    "inputSchema": {40                        "type": "object",41                        "properties": {}42                    }43                }44            ]45        }46    }47 48def handle_tools_call(params, req_id):49    tool_name = params.get("name")50    arguments = params.get("arguments", {})51 52    if tool_name == "echo":53        message = arguments.get("message", "")54        return {55            "jsonrpc": "2.0",56            "id": req_id,57            "result": {58                "content": [{"type": "text", "text": f"echo: {message}"}]59            }60        }61    elif tool_name == "crash":62        # Send a partial response then exit63        sys.stdout.write(json.dumps({"jsonrpc": "2.0", "id": req_id, "result": {"content": [{"type": "text", "text": "crashing..."}]}}) + "\n")64        sys.stdout.flush()65        os._exit(1)66    else:67        return {68            "jsonrpc": "2.0",69            "id": req_id,70            "error": {"code": -32602, "message": f"Unknown tool: {tool_name}"}71        }72 73HANDLERS = {74    "initialize": handle_initialize,75    "tools/list": handle_tools_list,76    "tools/call": handle_tools_call,77}78 79def main():80    sys.stdout = os.fdopen(sys.stdout.fileno(), "w", buffering=1)81    sys.stderr = os.fdopen(sys.stderr.fileno(), "w", buffering=1)82 83    for line in sys.stdin:84        line = line.strip()85        if not line:86            continue87        try:88            request = json.loads(line)89        except json.JSONDecodeError:90            continue91 92        method = request.get("method")93        req_id = request.get("id")94        params = request.get("params", {})95 96        # JSON-RPC 2.0: a message without an id is a notification and must not receive a response97        if req_id is None:98            continue99 100        handler = HANDLERS.get(method)101        if handler:102            response = handler(params, req_id)103        else:104            response = {105                "jsonrpc": "2.0",106                "id": req_id,107                "error": {"code": -32601, "message": f"Method not found: {method}"}108            }109 110        sys.stdout.write(json.dumps(response) + "\n")111        sys.stdout.flush()112 113if __name__ == "__main__":114    main()115