Model Context Provider (MCP) Server

import asyncio import json import logging from typing import Optional async def stdio_server(reader: asyncio.StreamReader, writer: asyncio.StreamWriter): """處理客戶端連接的伺服器函數""" try: peer = writer.get_extra_info('peername') logging.info(f"新的客戶端連接: {peer}") while True: # 讀取客戶端訊息 data = await reader.read(1024) if not data: break try: # 解析訊息 message = json.loads(data.decode()) # 處理訊息 response = { "status": "success", "message": "收到訊息" } # 發送回應 writer.write(json.dumps(response).encode()) await writer.drain() except json.JSONDecodeError: logging.error("無效的 JSON 格式") writer.write(json.dumps({ "status": "error", "message": "無效的訊息格式" }).encode()) await writer.drain() except Exception as e: logging.error(f"伺服器錯誤: {str(e)}") finally: writer.close() await writer.wait_closed() logging.info(f"客戶端斷開連接: {peer}")