QuantConnect MCP Server

Instructions

Implementation Reference

• quantconnect_mcp/src/tools/live_tools.py:144-252 (handler)

  The primary handler function for the 'read_live_algorithm' MCP tool. It authenticates with QuantConnect, makes a POST request to the 'live/read' API endpoint, parses the response, and returns comprehensive live algorithm details including status, runtime statistics, charts, and files.

  @mcp.tool()
  async def read_live_algorithm(
      project_id: int, deploy_id: Optional[str] = None
  ) -> Dict[str, Any]:
      """
      Read comprehensive live algorithm statistics, runtime data, and details.
  
      Args:
          project_id: ID of the project with the live algorithm
          deploy_id: Optional deploy ID for specific algorithm (omit to get latest)
  
      Returns:
          Dictionary containing detailed live algorithm statistics, runtime data, charts, and files
      """
      auth = get_auth_instance()
      if auth is None:
          return {
              "status": "error",
              "error": "QuantConnect authentication not configured. Use configure_auth() first.",
          }
  
      try:
          # Prepare request data
          request_data = {"projectId": project_id}
          if deploy_id:
              request_data["deployId"] = deploy_id
  
          # Make API request
          response = await auth.make_authenticated_request(
              endpoint="live/read", method="POST", json=request_data
          )
  
          # Parse response
          if response.status_code == 200:
              data = response.json()
  
              if data.get("success", False):
                  # Extract all the detailed information from LiveAlgorithmResults
                  deploy_id = data.get("deployId")
                  status = data.get("status")
                  message = data.get("message")
                  clone_id = data.get("cloneId")
                  launched = data.get("launched")
                  stopped = data.get("stopped")
                  brokerage = data.get("brokerage")
                  security_types = data.get("securityTypes")
                  project_name = data.get("projectName")
                  data_center = data.get("dataCenter")
                  public = data.get("public")
                  files = data.get("files", [])
                  runtime_statistics = data.get("runtimeStatistics", {})
                  charts = data.get("charts", {})
                  
                  return {
                      "status": "success",
                      "project_id": project_id,
                      "deploy_id": deploy_id,
                      "live_status": status,
                      "message": message,
                      "clone_id": clone_id,
                      "launched": launched,
                      "stopped": stopped,
                      "brokerage": brokerage,
                      "security_types": security_types,
                      "project_name": project_name,
                      "data_center": data_center,
                      "public": public,
                      "files": files,
                      "runtime_statistics": runtime_statistics,
                      "charts": charts,
                      "total_files": len(files),
                      "has_runtime_stats": bool(runtime_statistics),
                      "response": f"Successfully read live algorithm {deploy_id} for project {project_id}",
                  }
              else:
                  # API returned success=false
                  errors = data.get("errors", ["Unknown error"])
                  return {
                      "status": "error",
                      "error": "Failed to read live algorithm",
                      "details": errors,
                      "project_id": project_id,
                      "deploy_id": deploy_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: {str(e)}",
              "project_id": project_id,
              "deploy_id": deploy_id,
          }

• quantconnect_mcp/main.py:46-52 (registration)

  Registers the live_tools module (containing read_live_algorithm) by calling register_live_tools(mcp) during server initialization in the main entrypoint.

  safe_print("🔧 Registering QuantConnect tools...")
  register_auth_tools(mcp)
  register_project_tools(mcp)
  register_file_tools(mcp)
  register_backtest_tools(mcp)
  register_live_tools(mcp)
  register_optimization_tools(mcp)

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

  Alternative registration of live_tools module in the server.py module, called during server setup.

  safe_print("🔧 Registering QuantConnect tools...")
  register_auth_tools(mcp)
  register_project_tools(mcp)
  register_file_tools(mcp)
  register_backtest_tools(mcp)
  register_live_tools(mcp)
  register_optimization_tools(mcp)