QuantConnect MCP Server

Instructions

Implementation Reference

• quantconnect_mcp/src/tools/live_tools.py:11-142 (handler)

  The core handler function for the 'create_live_algorithm' tool. It validates inputs, prepares a request payload, authenticates using get_auth_instance(), and makes a POST request to QuantConnect's 'live/create' endpoint to deploy a live algorithm. Handles success/error responses and returns structured results.

  @mcp.tool()
  async def create_live_algorithm(
      project_id: int,
      compile_id: str,
      node_id: str,
      brokerage_id: str,
      brokerage_config: Dict[str, Any],
      data_providers: Optional[Dict[str, Any]] = None,
      version_id: str = "-1",
      parameters: Optional[Dict[str, Any]] = None,
      notifications: Optional[Dict[str, Any]] = None,
  ) -> Dict[str, Any]:
      """
      Create a live algorithm deployment.
  
      Args:
          project_id: ID of the project to deploy
          compile_id: Compile ID from successful project compilation
          node_id: ID of the node that will run the algorithm
          brokerage_id: Brokerage identifier (e.g., "QuantConnectBrokerage", "InteractiveBrokersBrokerage")
          brokerage_config: Brokerage configuration dictionary with credentials and settings
          data_providers: Optional data provider configurations (defaults to same as brokerage)
          version_id: Version of Lean to use (default: "-1" for master)
          parameters: Optional algorithm parameters
          notifications: Optional notification settings
  
      Returns:
          Dictionary containing live algorithm deployment result
      """
      auth = get_auth_instance()
      if auth is None:
          return {
              "status": "error",
              "error": "QuantConnect authentication not configured. Use configure_auth() first.",
          }
  
      # Validate required brokerage config
      if not isinstance(brokerage_config, dict) or "id" not in brokerage_config:
          return {
              "status": "error",
              "error": "brokerage_config must be a dictionary with 'id' field",
          }
  
      try:
          # Prepare request data
          request_data = {
              "versionId": version_id,
              "projectId": project_id,
              "compileId": compile_id,
              "nodeId": node_id,
              "brokerage": brokerage_config,
          }
  
          # Set up data providers (default to same as brokerage if not specified)
          if data_providers is None:
              request_data["dataProviders"] = {
                  brokerage_id: {"id": brokerage_id}
              }
          else:
              request_data["dataProviders"] = data_providers
  
          # Add optional parameters
          if parameters:
              request_data["parameters"] = parameters
          else:
              request_data["parameters"] = {}
  
          if notifications:
              request_data["notification"] = notifications
          else:
              request_data["notification"] = {}
  
          # Make API request
          response = await auth.make_authenticated_request(
              endpoint="live/create", method="POST", json=request_data
          )
  
          # Parse response
          if response.status_code == 200:
              data = response.json()
  
              if data.get("success", False):
                  live_algorithm = data.get("live", {})
                  deploy_id = live_algorithm.get("deployId")
                  status = live_algorithm.get("status")
  
                  return {
                      "status": "success",
                      "project_id": project_id,
                      "compile_id": compile_id,
                      "deploy_id": deploy_id,
                      "live_status": status,
                      "brokerage": live_algorithm.get("brokerage"),
                      "launched": live_algorithm.get("launched"),
                      "live_algorithm": live_algorithm,
                      "message": f"Successfully created live algorithm {deploy_id} with status: {status}",
                  }
              else:
                  # API returned success=false
                  errors = data.get("errors", ["Unknown error"])
                  return {
                      "status": "error",
                      "error": "Live algorithm creation failed",
                      "details": errors,
                      "project_id": project_id,
                      "compile_id": compile_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 create live algorithm: {str(e)}",
              "project_id": project_id,
              "compile_id": compile_id,
          }

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

  Calls register_live_tools(mcp) to register the create_live_algorithm tool (and other live tools) with the MCP server during initialization.

  register_live_tools(mcp)

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

  Alternative entry point that calls register_live_tools(mcp) to register the live tools including create_live_algorithm.

  register_live_tools(mcp)

• quantconnect_mcp/src/tools/live_tools.py:8-8 (registration)

  The registration function that defines and registers the create_live_algorithm tool via @mcp.tool() decorator.

  def register_live_tools(mcp: FastMCP):