QuantConnect MCP Server

Instructions

Implementation Reference

• quantconnect_mcp/src/tools/project_tools.py:934-1013 (handler)

  The main execution logic for the 'update_project_nodes' tool. Decorated with @mcp.tool(), it authenticates with QuantConnect, prepares a POST request to the 'projects/nodes/update' endpoint with project_id and nodes dict, handles responses, and returns success/error dicts.

  @mcp.tool()
  async def update_project_nodes(
      project_id: int, nodes: Dict[str, bool]
  ) -> Dict[str, Any]:
      """
      Update the active state of the given nodes to true.
  
      Args:
          project_id: ID of the project to update nodes for
          nodes: Dictionary mapping node IDs to their active state (true/false)
  
      Returns:
          Dictionary containing update result
      """
      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,
              "nodes": nodes,
          }
  
          # Make API request
          response = await auth.make_authenticated_request(
              endpoint="projects/nodes/update", method="POST", json=request_data
          )
  
          # Parse response
          if response.status_code == 200:
              data = response.json()
  
              if data.get("success", False):
                  active_nodes = [node_id for node_id, active in nodes.items() if active]
                  
                  return {
                      "status": "success",
                      "project_id": project_id,
                      "updated_nodes": nodes,
                      "active_nodes": active_nodes,
                      "message": f"Successfully updated {len(nodes)} node(s) for project {project_id}, {len(active_nodes)} now active",
                  }
              else:
                  # API returned success=false
                  errors = data.get("errors", ["Unknown error"])
                  return {
                      "status": "error",
                      "error": "Failed to update project nodes",
                      "details": errors,
                      "project_id": project_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 update project nodes: {str(e)}",
              "project_id": project_id,
          }

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

  Registration of project tools (including 'update_project_nodes') by calling register_project_tools(mcp) during server initialization.

  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/main.py:46-52 (registration)

  Alternative entrypoint registration of project tools (including 'update_project_nodes') by calling register_project_tools(mcp).

  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/tools/project_tools.py:935-947 (schema)

  Type hints defining the input schema: project_id (int), nodes (Dict[str, bool]) and return Dict[str, Any].

  async def update_project_nodes(
      project_id: int, nodes: Dict[str, bool]
  ) -> Dict[str, Any]:
      """
      Update the active state of the given nodes to true.
  
      Args:
          project_id: ID of the project to update nodes for
          nodes: Dictionary mapping node IDs to their active state (true/false)
  
      Returns:
          Dictionary containing update result
      """