修改grpc

01005-发送消息-chat write.robot

@@ -386,3 +386,12 @@ ${xlsx_file}      testfile.xlsx
     log    ${content}
     Should Be Equal As Strings    ${content["code"]}    500
     Should Be Equal As Strings    ${content["message"]}    Server busy, please try later
+
+正例-正常发送文字消息，进入探索
+    [Tags]    F
+    create session    URI    ${URL}
+    ${content}    POST请求结果_PARAMS    ${path}    {"sessionId":"${sessionId}","text":"使用探索工具，打开百度"}    200    ${token}
+    log    ${content}
+    Should Be Equal As Strings    ${content["code"]}    200
+    Should Be Equal As Strings    ${content["message"]}    成功
+    Set Global Variable    ${ex_streamId}    ${content["data"]}

grpc_manager/main.py

 import asyncio
 import logging
-
 import requests
 import json
 from grpc_client import GrpcClient
@@ -10,79 +9,122 @@ from result import ResultModel
 # 配置日志
 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
 
-# 定义消息处理函数
-async def handle_message(request_path: str, request_data: str) -> ResultModel:
-    print(f"Received request on path: {request_path}, data: {request_data}")
-    return ResultModel(code=200, data={"message": "Request processed successfully"})
+
+async def handle_message(request_path: str, request_data: str,request_id:str) -> ResultModel:
+    """增强版消息处理函数"""
+    logging.info(f"收到请求路径: {request_path}, 数据: {request_data}")
+
+    # 特定消息处理逻辑
+    if request_path == "list_mcp_servers":  # 特定指令1
+        return ResultModel(code=200, data=[])
+    elif request_path == "adaptable_activities":  # 特定指令2
+        return ResultModel(code=200, data=[  {"id": "sleep_seconds", "name": "sleep_seconds", "displayName": "等待", "description": "此工具的用途是在目标无法达成时进行等待，最多等待2次", "parametersSchema": {"description": "等待", "properties": {"wait_seconds": {"default": 5, "description": "默认5秒，可修改为1-5秒时间", "title": "Wait Seconds", "type": "integer"}}, "title": "SleepSeconds", "type": "object", "required": []}, "elementIdJsonPaths": [], "icon": "sleep_seconds", "timeout": 180}])
+
+    # 默认响应
+    return ResultModel(code=200, data={"message": "请求处理成功"})
+
 
 async def send_ping(listener: GrpcListener):
-    """定时发送 ping 消息"""
+    """心跳维护任务"""
     while True:
         try:
             await listener.send_to_all_stream("ping", "#ping", type=-1, retry_count=3)
+            logging.info("心跳包已发送")
+        except Exception as e:
+            logging.error(f"发送心跳异常: {e}")
+        await asyncio.sleep(5)  # 每5秒发送一次
+
+
+async def respond_to_server(listener: GrpcListener):
+    """服务端消息响应器"""
+    while True:
+        try:
+            # 实际业务中可添加消息检测逻辑
+            # 示例：当收到特定通知时发送响应
+            # if 收到特殊通知:
+            #     response = {"status": "OK", "timestamp": int(time.time())}
+            #     await listener.send_to_all_stream("response", json.dumps(response))
+
+            await asyncio.sleep(1)  # 防止CPU过载
         except Exception as e:
-            logging.error(f"send ping error: {e}")
-        await asyncio.sleep(5)  # 每 5 秒发送一次 ping
+            logging.error(f"响应器异常: {e}")
+            await asyncio.sleep(5)  # 异常时延长等待
+
 
 async def main():
-    headers = {
-        "Content-Type" : "application/x-www-form-urlencoded",
-        "Sm4-Key" : "dUVNkTJLlzMHPqBTZh085jT0N6S7930l6sgddJsYSC8=",
-        "Cookie" : "ENCRYPT_KEY=f48bf3-05fe-4978"
-    }
-    params = {"username":"api_tester",
-              "password":"$SM4$Nd/e40vIS6UD9VMaSOmYGlrBORfo7r5/1z9D5d4E4es=$4MS$",
-              "tenantCode":"API_test",
-              "scope":"all",
-              "client_id":"a68ad587830d41aebf418a919006353e",
-              "client_secret":"Ninetech@123",
-              "grant_type":"password"
-    }
-    res = requests.post(url="http://accounts-4j.sit.ninetechone.com/oauth/token",headers=headers,params=params)
-    res1 = requests.get("http://bitagent.sit.ninetechone.com/api/system/config/client")
-    grpc = json.loads(res1.json()["data"])
-    # gRPC 配置
-    config = {
-        "token": "Bearer " + res.json()["access_token"],
-        "client_id": "a68ad587830d41aebf418a919006353e",
-        "host": grpc["grpc_host"],
-        "is_ssl": grpc["grpc_use_ssl"]
+    # 认证配置
+    auth_config = {
+        "headers": {
+            "Content-Type": "application/x-www-form-urlencoded",
+            "Sm4-Key": "dUVNkTJLlzMHPqBTZh085jT0N6S7930l6sgddJsYSC8=",
+            "Cookie": "ENCRYPT_KEY=f48bf3-05fe-4978"
+        },
+        "params": {
+            "username": "api_tester",
+            "password": "$SM4$Nd/e40vIS6UD9VMaSOmYGlrBORfo7r5/1z9D5d4E4es=$4MS$",
+            "tenantCode": "API_test",
+            "scope": "all",
+            "client_id": "a68ad587830d41aebf418a919006353e",
+            "client_secret": "Ninetech@123",
+            "grant_type": "password"
+        }
     }
-    print(config)
 
-    # 创建 GrpcClient 实例
-    client = GrpcClient()
-    # 设置头部信息
-    client.set_header(config["token"], config["client_id"])
+    try:
+        # 获取访问令牌
+        token_res = requests.post(
+            url="http://accounts-4j.sit.ninetechone.com/oauth/token",
+            headers=auth_config["headers"],
+            params=auth_config["params"]
+        )
+        token_res.raise_for_status()
 
-    # 连接到 gRPC 服务器
-    connected = await client.connect(config["host"], config["is_ssl"])
-    if not connected:
-        logging.error("Failed to connect to the gRPC server.")
-        return
+        # 获取gRPC配置
+        config_res = requests.get("http://bitagent.sit.ninetechone.com/api/system/config/client")
+        config_res.raise_for_status()
 
-    # 创建 GrpcListener 实例
-    listener = GrpcListener(client)
-    # 设置消息处理函数
-    listener.message_received = handle_message
-    # 启动监听
-    listener.start()
+        grpc_config = {
+            "token": "Bearer " + token_res.json()["access_token"],
+            "client_id": auth_config["params"]["client_id"],
+            "host": json.loads(config_res.json()["data"])["grpc_host"],
+            "is_ssl": json.loads(config_res.json()["data"])["grpc_use_ssl"]
+        }
+        logging.info(f"gRPC配置加载完成: {grpc_config}")
 
-    # 启动发送 ping 任务
-    ping_task = asyncio.create_task(send_ping(listener))
+        # 初始化gRPC客户端
+        client = GrpcClient()
+        client.set_header(grpc_config["token"], grpc_config["client_id"])
 
-    try:
-        # 保持程序运行
-        while True:
-            await asyncio.sleep(1)
-    except KeyboardInterrupt:
-        logging.info("Shutting down...")
-        # 停止监听
-        await listener.stop()
-        # 断开连接
-        await client.disconnect()
-        # 取消 ping 任务
-        ping_task.cancel()
+        # 建立连接
+        if not await client.connect(grpc_config["host"], grpc_config["is_ssl"]):
+            logging.error("gRPC服务器连接失败")
+            return
+
+        # 启动消息监听器
+        listener = GrpcListener(client)
+        listener.message_received = handle_message
+        listener.start()
+
+        # 启动后台任务
+        tasks = [
+            asyncio.create_task(send_ping(listener)),
+            asyncio.create_task(respond_to_server(listener))
+        ]
+
+        # 主循环
+        try:
+            while True:
+                await asyncio.sleep(1)
+        except KeyboardInterrupt:
+            logging.info("正在关闭服务...")
+            await listener.stop()
+            await client.disconnect()
+            for task in tasks:
+                task.cancel()
+            await asyncio.gather(*tasks, return_exceptions=True)
+
+    except Exception as e:
+        logging.error(f"服务启动失败: {e}")
 
 
 if __name__ == "__main__":