QuantConnect MCP Server

Instructions

Implementation Reference

• quantconnect_mcp/src/tools/file_tools.py:88-184 (handler)

  The core handler implementation for the 'read_file' MCP tool. This async function handles reading specific files or all files from a QuantConnect project via authenticated API calls to 'files/read'. Includes input validation via type hints, comprehensive error handling, and structured JSON responses. Registered directly via @mcp.tool() decorator.

  @mcp.tool()
  async def read_file(project_id: int, name: Optional[str] = None) -> Dict[str, Any]:
      """
      Read a specific file from a project or all files if no name provided.
  
      Args:
          project_id: ID of the project to read files from
          name: Optional name of specific file to read. If not provided, reads all files.
  
      Returns:
          Dictionary containing file content(s) or error information
      """
      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: Dict[str, Any] = {"projectId": project_id}
          if name is not None:
              request_data["name"] = name
  
          # Make API request
          response = await auth.make_authenticated_request(
              endpoint="files/read", method="POST", json=request_data
          )
  
          # Parse response
          if response.status_code == 200:
              data = response.json()
  
              if data.get("success", False):
                  files = data.get("files", [])
  
                  # If specific file was requested
                  if name is not None:
                      if files:
                          file_data = files[0]
                          return {
                              "status": "success",
                              "project_id": project_id,
                              "file": file_data,
                              "message": f"Successfully read file '{name}' from project {project_id}",
                          }
                      else:
                          return {
                              "status": "error",
                              "error": f"File '{name}' not found in project {project_id}",
                          }
  
                  # If all files were requested
                  else:
                      return {
                          "status": "success",
                          "project_id": project_id,
                          "files": files,
                          "total_files": len(files),
                          "message": f"Successfully read {len(files)} files from project {project_id}",
                      }
              else:
                  # API returned success=false
                  errors = data.get("errors", ["Unknown error"])
                  return {
                      "status": "error",
                      "error": "File read failed",
                      "details": errors,
                      "project_id": project_id,
                      "file_name": name,
                  }
  
          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 file(s): {str(e)}",
              "project_id": project_id,
              "file_name": name,
          }

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

  Server initialization calls register_file_tools(mcp), which defines and registers the read_file tool (along with other file tools) using FastMCP's @mcp.tool() decorators.

  register_file_tools(mcp)

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

  Entry point script also registers file tools, importing mcp from server.py, providing an alternative invocation path.

  register_file_tools(mcp)