QuantConnect MCP Server

Instructions

Implementation Reference

• quantconnect_mcp/src/tools/live_tools.py:485-604 (handler)

  The handler function that implements the 'read_live_logs' tool. It handles authentication, input validation for log line ranges, constructs the API request to QuantConnect's live/logs/read endpoint, and parses the response to return log data or error information.

  @mcp.tool()
  async def read_live_logs(
      project_id: int,
      algorithm_id: str,
      start_line: int,
      end_line: int,
      format: str = "json",
  ) -> Dict[str, Any]:
      """
      Read logs from a live algorithm.
  
      Args:
          project_id: Project ID of the live running algorithm
          algorithm_id: Deploy ID (Algorithm ID) of the live running algorithm
          start_line: Start line of logs to read
          end_line: End line of logs to read (difference must be < 250)
          format: Format of log results (default: "json")
  
      Returns:
          Dictionary containing live algorithm logs
      """
      auth = get_auth_instance()
      if auth is None:
          return {
              "status": "error",
              "error": "QuantConnect authentication not configured. Use configure_auth() first.",
          }
  
      # Validate line range
      if end_line <= start_line:
          return {
              "status": "error",
              "error": "end_line must be greater than start_line",
          }
  
      if end_line - start_line >= 250:
          return {
              "status": "error",
              "error": "Line range too large: difference between start_line and end_line must be less than 250",
          }
  
      if start_line < 0 or end_line < 0:
          return {
              "status": "error",
              "error": "start_line and end_line must be non-negative",
          }
  
      try:
          # Prepare request data
          request_data = {
              "format": format,
              "projectId": project_id,
              "algorithmId": algorithm_id,
              "startLine": start_line,
              "endLine": end_line,
          }
  
          # Make API request
          response = await auth.make_authenticated_request(
              endpoint="live/logs/read", method="POST", json=request_data
          )
  
          # Parse response
          if response.status_code == 200:
              data = response.json()
  
              if data.get("success", False):
                  logs = data.get("logs", [])
                  length = data.get("length", 0)
                  deployment_offset = data.get("deploymentOffset", 0)
                  
                  return {
                      "status": "success",
                      "project_id": project_id,
                      "algorithm_id": algorithm_id,
                      "start_line": start_line,
                      "end_line": end_line,
                      "logs": logs,
                      "length": length,
                      "deployment_offset": deployment_offset,
                      "format": format,
                      "message": f"Successfully retrieved {len(logs)} log lines from live algorithm {algorithm_id}",
                  }
              else:
                  # API returned success=false
                  errors = data.get("errors", ["Unknown error"])
                  return {
                      "status": "error",
                      "error": "Failed to read live algorithm logs",
                      "details": errors,
                      "project_id": project_id,
                      "algorithm_id": algorithm_id,
                  }
  
          elif response.status_code == 401:
              return {
                  "status": "error",
                  "error": "Authentication failed. Check your credentials and ensure they haven't expired.",
              }
  
          else:
              return {
                  "status": "error",
                  "error": f"API request failed with status {response.status_code}",
                  "response_text": (
                      response.text[:500]
                      if hasattr(response, "text")
                      else "No response text"
                  ),
              }
  
      except Exception as e:
          return {
              "status": "error",
              "error": f"Failed to read live algorithm logs: {str(e)}",
              "project_id": project_id,
              "algorithm_id": algorithm_id,
              "start_line": start_line,
              "end_line": end_line,
          }

• quantconnect_mcp/main.py:51-51 (registration)

  Call to register_live_tools(mcp) which registers all live trading tools, including 'read_live_logs', with the MCP server instance.

  register_live_tools(mcp)

• quantconnect_mcp/src/server.py:78-78 (registration)

  Call to register_live_tools(mcp) in the server initialization, registering the live tools including 'read_live_logs'.

  register_live_tools(mcp)