Qinglong MCP Server

Instructions

Implementation Reference

• server.py:247-294 (handler)

  The handler for the 'get_task_logs' tool. It retrieves the specified task's cron details from the Qinglong API, then fetches and returns the execution logs if available.

  elif tool_name == "get_task_logs":
      task_id = arguments.get("task_id")
      try:
          url = f"{QINGLONG_URL}/open/crons/{task_id}"
          headers = {"Authorization": f"Bearer {token}"}
          resp = requests.get(url, headers=headers, timeout=10)
          result = resp.json()
      except Exception as e:
          response = {
              "jsonrpc": "2.0",
              "id": request["id"],
              "error": {"code": -32603, "message": f"请求失败: {str(e)}"}
          }
          print(json.dumps(response), flush=True)
          continue
      
      if result.get("code") == 200:
          cron = result["data"]
          log_path = cron.get("log_path", "")
          
          if log_path:
              try:
                  log_url = f"{QINGLONG_URL}/open/crons/{task_id}/log"
                  log_resp = requests.get(log_url, headers=headers, timeout=10)
                  log_result = log_resp.json()
                  
                  if log_result.get("code") == 200:
                      log_content = log_result["data"]
                      output = f"任务 {task_id} ({cron.get('name')}) 的执行日志:\n\n{log_content}"
                  else:
                      output = f"获取日志失败: {log_result}"
              except Exception as e:
                  output = f"读取日志失败: {str(e)}"
          else:
              output = f"任务 {task_id} ({cron.get('name')}) 暂无执行日志"
          
          response = {
              "jsonrpc": "2.0",
              "id": request["id"],
              "result": {"content": [{"type": "text", "text": output}]}
          }
      else:
          response = {
              "jsonrpc": "2.0",
              "id": request["id"],
              "error": {"code": -32603, "message": f"获取任务信息失败: {result}"}
          }

• server.py:112-122 (schema)

  The input schema for the 'get_task_logs' tool, defining the required 'task_id' parameter as an integer.

  {
      "name": "get_task_logs",
      "description": "获取青龙面板中指定任务的执行日志",
      "inputSchema": {
          "type": "object",
          "properties": {
              "task_id": {"type": "integer", "description": "任务 ID"}
          },
          "required": ["task_id"]
      }
  },

• server.py:77-156 (registration)

  The tool registration in the 'tools/list' method response, where 'get_task_logs' is listed among available tools with its schema.

  response = {
      "jsonrpc": "2.0",
      "id": request["id"],
      "result": {
          "tools": [
              {
                  "name": "list_qinglong_tasks",
                  "description": "查询青龙面板中的所有定时任务列表",
                  "inputSchema": {
                      "type": "object",
                      "properties": {}
                  }
              },
              {
                  "name": "run_task",
                  "description": "执行任务并等待完成，返回执行日志（最多等待30秒）",
                  "inputSchema": {
                      "type": "object",
                      "properties": {
                          "task_id": {"type": "integer", "description": "任务 ID"}
                      },
                      "required": ["task_id"]
                  }
              },
              {
                  "name": "run_task_async",
                  "description": "异步启动任务，不等待执行完成",
                  "inputSchema": {
                      "type": "object",
                      "properties": {
                          "task_id": {"type": "integer", "description": "任务 ID"}
                      },
                      "required": ["task_id"]
                  }
              },
              {
                  "name": "get_task_logs",
                  "description": "获取青龙面板中指定任务的执行日志",
                  "inputSchema": {
                      "type": "object",
                      "properties": {
                          "task_id": {"type": "integer", "description": "任务 ID"}
                      },
                      "required": ["task_id"]
                  }
              },
              {
                  "name": "get_task_status",
                  "description": "获取青龙面板中指定任务的执行状态",
                  "inputSchema": {
                      "type": "object",
                      "properties": {
                          "task_id": {"type": "integer", "description": "任务 ID"}
                      },
                      "required": ["task_id"]
                  }
              },
              {
                  "name": "list_subscriptions",
                  "description": "查询青龙面板中的所有订阅列表",
                  "inputSchema": {
                      "type": "object",
                      "properties": {}
                  }
              },
              {
                  "name": "run_subscription",
                  "description": "运行指定的订阅",
                  "inputSchema": {
                      "type": "object",
                      "properties": {
                          "subscription_id": {"type": "integer", "description": "订阅 ID"}
                      },
                      "required": ["subscription_id"]
                  }
              }
          ]
      }
  }
  print(json.dumps(response), flush=True)