import json
from http.server import BaseHTTPRequestHandler, HTTPServer
# 定義一個(gè)Tool類，用于存儲(chǔ)工具的元數(shù)據(jù)和執(zhí)行函數(shù)
class Tool:
    def __init__(self, name, description, input_schema, output_schema, func):
        self.name = name
        self.description = description
        self.input_schema = input_schema    # JSON Schema for input parameters
        self.output_schema = output_schema  # JSON Schema for output
        self.func = func                   # Function to execute the tool
# MCP服務(wù)器類，維護(hù)工具和資源，并處理JSON-RPC請(qǐng)求
class MCPServer:
    def __init__(self):
        # 注冊(cè)工具列表：這里添加一個(gè)“計(jì)算器”工具，提供加法和乘法 功能
        def add(a, b):  # 簡(jiǎn)單加法函數(shù)
            return a + b
        def multiply(a, b):  # 簡(jiǎn)單乘法函數(shù)
            return a * b
        # 為演示，將加法和乘法作為兩個(gè)獨(dú)立工具注冊(cè)
        self.tools = {
            "add": Tool(
                name="add",
                description="Add two numbers",
                input_schema={  # 輸入?yún)?shù)的JSON Schema定義
                    "type": "object",
                    "properties": {
                        "a": {"type": "number", "description": "First operand"},
                        "b": {"type": "number", "description": "Second operand"}
                    },
                    "required": ["a", "b"]
                },
                output_schema={  # 輸出結(jié)果的JSON Schema定義
                    "type": "number"
                },
                func=add
            ),
            "multiply": Tool(
                name="multiply",
                description="Multiply two numbers",
                input_schema={
                    "type": "object",
                    "properties": {
                        "a": {"type": "number", "description": "First operand"},
                        "b": {"type": "number", "description": "Second operand"}
                    },
                    "required": ["a", "b"]
                },
                output_schema={
                    "type": "number"
                },
                func=multiply
            )
        }
        # 可選：注冊(cè)資源列表，這里添加一個(gè)簡(jiǎn)單文本資源示例
        self.resources = {
            "mem://greeting": {
                "name": "greeting.txt",
                "title": "Greeting Message",
                "description": "A welcome text message",
                "mimeType": "text/plain",
                "text": "Hello, welcome to the MCP demo!"  # 資源內(nèi)容
            }
        }
    def handle_request(self, request):
        """
        處理JSON-RPC請(qǐng)求，根據(jù)method調(diào)用對(duì)應(yīng)的處理邏輯。
        返回JSON可序列化的響應(yīng)字典。
        """
        # 基礎(chǔ)的JSON-RPC字段解析
        jsonrpc = request.get("jsonrpc")
        req_id = request.get("id")
        method = request.get("method")
        params = request.get("params", {})
        # 確保符合 JSON-RPC 2.0 協(xié)議
        if jsonrpc != "2.0" or req_id is None or method is None:
            # 返回 JSON-RPC 錯(cuò)誤：-32600 = Invalid Request
            return {
                "jsonrpc": "2.0",
                "id": req_id,
                "error": {"code": -32600, "message": "Invalid Request"}
            }
        # 處理方法：tools/list, tools/call, resources/list, resources/read
        try:
            if method == "tools/list":
                # 列出可用工具
                tools_info = []
                for tool in self.tools.values():
                    tools_info.append({
                        "name": tool.name,
                        "description": tool.description,
                        "inputSchema": tool.input_schema,
                        "outputSchema": tool.output_schema
                    })
                return {
                    "jsonrpc": "2.0",
                    "id": req_id,
                    "result": {
                        "tools": tools_info
                    }
                }
            elif method == "tools/call":
                # 調(diào)用指定名稱的工具
                name = params.get("name")
                arguments = params.get("arguments", {})
                if not name or name not in self.tools:
                    # 工具名稱不存在
                    return {
                        "jsonrpc": "2.0",
                        "id": req_id,
                        "error": {"code": -32601, "message": f"Tool '{name}' not found"}
                    }
                # 獲取工具并執(zhí)行
                tool = self.tools[name]
                try:
                    # 調(diào)用工具函數(shù)，傳入?yún)?shù)
                    result_value = tool.func(**arguments)
                except TypeError as e:
                    # 參數(shù)不正確或缺失
                    return {
                        "jsonrpc": "2.0",
                        "id": req_id,
                        "error": {"code": -32602, "message": f"Invalid parameters: {str(e)}"}
                    }
                except Exception as e:
                    # 工具執(zhí)行過(guò)程中出現(xiàn)其他異常
                    return {
                        "jsonrpc": "2.0",
                        "id": req_id,
                        "error": {"code": -32000, "message": f"Tool execution error: {str(e)}"}
                    }
                # 返回執(zhí)行結(jié)果（這里結(jié)果直接是數(shù)值，符合 outputSchema 定義）
                return {
                    "jsonrpc": "2.0",
                    "id": req_id,
                    "result": {
                        "output": result_value  # 將結(jié)果包裝在一個(gè)字段中
                    }
                }
            elif method == "resources/list":
                # 列出可用資源
                resources_info = []
                for uri, res in self.resources.items():
                    resources_info.append({
                        "uri": uri,
                        "name": res.get("name"),
                        "title": res.get("title"),
                        "description": res.get("description"),
                        "mimeType": res.get("mimeType")
                    })
                return {
                    "jsonrpc": "2.0",
                    "id": req_id,
                    "result": {
                        "resources": resources_info
                    }
                }
            elif method == "resources/read":
                # 讀取指定URI的資源內(nèi)容
                uri = params.get("uri")
                if not uri or uri not in self.resources:
                    return {
                        "jsonrpc": "2.0",
                        "id": req_id,
                        "error": {"code": -32602, "message": f"Invalid resource URI: {uri}"}
                    }
                res = self.resources[uri]
                # 返回資源內(nèi)容（文本）
                return {
                    "jsonrpc": "2.0",
                    "id": req_id,
                    "result": {
                        "contents": [
                            {
                                "uri": uri,
                                "name": res.get("name"),
                                "title": res.get("title"),
                                "mimeType": res.get("mimeType"),
                                "text": res.get("text")
                            }
                        ]
                    }
                }
            else:
                # 未知的方法
                return {
                    "jsonrpc": "2.0",
                    "id": req_id,
                    "error": {"code": -32601, "message": f"Method '{method}' not found"}
                }
        except Exception as e:
            # 捕獲任何未預(yù)料的異常，返回內(nèi)部錯(cuò)誤
            return {
                "jsonrpc": "2.0",
                "id": req_id,
                "error": {"code": -32603, "message": f"Internal error: {str(e)}"}
            }
# 實(shí)例化 MCPServer，全局使用
mcp_server = MCPServer()
# 定義HTTP請(qǐng)求處理器類
class MCPRequestHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        # 讀取請(qǐng)求體中的JSON數(shù)據(jù)
        content_length = int(self.headers.get('Content-Length', 0))
        request_bytes = self.rfile.read(content_length)
        try:
            request_json = json.loads(request_bytes.decode('utf-8'))
        except json.JSONDecodeError:
            # 如果不是合法的JSON，返回錯(cuò)誤
            response = {
                "jsonrpc": "2.0",
                "id": None,
                "error": {"code": -32700, "message": "Parse error"}
            }
            self._send_response(response)
            return
        # 調(diào)用MCPServer處理請(qǐng)求，獲取響應(yīng)
        response = mcp_server.handle_request(request_json)
        # 將響應(yīng)發(fā)送回客戶端
        self._send_response(response)
    def _send_response(self, response_obj):
        """輔助方法：發(fā)送JSON響應(yīng)。"""
        response_bytes = json.dumps(response_obj).encode('utf-8')
        # 構(gòu)造HTTP響應(yīng)頭
        self.send_response(200)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(response_bytes)))
        self.end_headers()
        # 寫(xiě)出響應(yīng)內(nèi)容
        self.wfile.write(response_bytes)
        self.wfile.flush()
# 啟動(dòng) HTTP 服務(wù)器（監(jiān)聽(tīng) localhost:8000）
if __name__ == "__main__":
    server_address = ('', 8000)  # 監(jiān)聽(tīng)在所有接口的8000端口
    httpd = HTTPServer(server_address, MCPRequestHandler)
    print("MCP Server is running at http://localhost:8000")
    print("Available tools:", [name for name in mcp_server.tools.keys()])
    httpd.serve_forever()

import json
import http.client

# MCP客戶端輔助函數(shù)：發(fā)送JSON-RPC請(qǐng)求并返回解析后的結(jié)果或錯(cuò)誤
def send_json_rpc(method, params=None, request_id=1):
    conn = http.client.HTTPConnection("localhost", 8000)
    request_obj = {
        "jsonrpc": "2.0",
        "id": request_id,
        "method": method,
        "params": params or {}
    }
    # 將請(qǐng)求對(duì)象序列化為JSON并發(fā)送
    conn.request("POST", "/", body=json.dumps(request_obj), headers={
        "Content-Type": "application/json"
    })
    # 獲取HTTP響應(yīng)并解析JSON
    response = conn.getresponse()
    data = response.read().decode('utf-8')
    conn.close()
    try:
        return json.loads(data)
    except json.JSONDecodeError:
        print("Invalid JSON response:", data)
        return None

# 1. 發(fā)現(xiàn)可用工具列表
resp = send_json_rpc("tools/list", {}, request_id=1)
print("Available tools:")
if "result" in resp:
    for tool in resp["result"]["tools"]:
        name = tool["name"]
        desc = tool.get("description")
        print(f" - {name}: {desc}")
else:
    # 錯(cuò)誤情況
    print("Error listing tools:", resp.get("error"))

# 2. 調(diào)用加法工具 (例如 3 + 5)
add_params = {"name": "add", "arguments": {"a": 3, "b": 5}}
resp2 = send_json_rpc("tools/call", add_params, request_id=2)
if "result" in resp2:
    result_value = resp2["result"].get("output")
    print(f"Result of add(3, 5): {result_value}")
else:
    print("Error invoking add tool:", resp2.get("error"))

# 3. 調(diào)用乘法工具 (例如 4 * 7)
mul_params = {"name": "multiply", "arguments": {"a": 4, "b": 7}}
resp3 = send_json_rpc("tools/call", mul_params, request_id=3)
if "result" in resp3:
    result_value = resp3["result"].get("output")
    print(f"Result of multiply(4, 7): {result_value}")
else:
    print("Error invoking multiply tool:", resp3.get("error"))

# 4. （可選）列出可用資源
resp4 = send_json_rpc("resources/list", {}, request_id=4)
print("Available resources:")
if "result" in resp4:
    for res in resp4["result"]["resources"]:
        uri = res["uri"]
        desc = res.get("description")
        print(f" - {uri}: {desc}")
else:
    print("Error listing resources:", resp4.get("error"))

# 5. （可選）讀取指定資源內(nèi)容 (例如 mem://greeting)
res_params = {"uri": "mem://greeting"}
resp5 = send_json_rpc("resources/read", res_params, request_id=5)
if "result" in resp5:
    contents = resp5["result"].get("contents", [])
    if contents:
        text = contents[0].get("text")
        print(f"Content of resource mem://greeting: {text}")
else:
    print("Error reading resource:", resp5.get("error"))

Available tools:
 - add: Add two numbers
 - multiply: Multiply two numbers
Result of add(3, 5): 8
Result of multiply(4, 7): 28
Available resources:
 - mem://greeting: A welcome text message
Content of resource mem://greeting: Hello, welcome to the MCP demo!

python mcp_server.py

MCP Server is running at http://localhost:8000
Available tools: ['add', 'multiply']

python mcp_client.py

Available tools:
 - add: Add two numbers
 - multiply: Multiply two numbers
Result of add(3, 5): 8
Result
 
of multiply(4, 7): 28
Available resources:
 - mem://greeting: A welcome text message
Content of resource mem://greeting: Hello, welcome to the MCP demo!

curl -X POST http://localhost:8000/ -H "Content-Type: application/json" \
     -d '{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}'