QuantConnect MCP Server

read_live_portfolio

Retrieve real-time portfolio data from live trading algorithms to monitor positions, holdings, and current market value for informed decision-making.

Instructions

Read portfolio state from a live algorithm.

Args: project_id: Project ID of the live algorithm

Returns: Dictionary containing live algorithm portfolio state

Input Schema

TableJSON Schema

Name	Required	Description	Default
`project_id`	Yes

Output Schema

TableJSON Schema

Name	Required	Description	Default
No arguments

Implementation Reference

quantconnect_mcp/src/tools/live_tools.py:727-798 (handler)

Handler function that implements the read_live_portfolio tool. It authenticates with QuantConnect, makes an API request to /live/portfolio/read, and returns the portfolio state or error details.

async def read_live_portfolio(project_id: int) -> Dict[str, Any]:
    """
    Read portfolio state from a live algorithm.

    Args:
        project_id: Project ID of the live algorithm

    Returns:
        Dictionary containing live algorithm portfolio state
    """
    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}

        # Make API request
        response = await auth.make_authenticated_request(
            endpoint="live/portfolio/read", method="POST", json=request_data
        )

        # Parse response
        if response.status_code == 200:
            data = response.json()

            if data.get("success", False):
                portfolio = data.get("portfolio", {})
                
                return {
                    "status": "success",
                    "project_id": project_id,
                    "portfolio": portfolio,
                    "message": f"Successfully retrieved portfolio state from live algorithm {project_id}",
                }
            else:
                # API returned success=false
                errors = data.get("errors", ["Unknown error"])
                return {
                    "status": "error",
                    "error": "Failed to read live algorithm portfolio",
                    "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 read live algorithm portfolio: {str(e)}",
            "project_id": project_id,
        }

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

The register_live_tools function defines and registers all live trading tools, including read_live_portfolio via @mcp.tool() decorator inside it. This function is called from main.py and server.py to register the tools with the MCP server.

def register_live_tools(mcp: FastMCP):
    """Register live trading tools with the MCP server."""

    @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,
            }

    @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,
            }

    @mcp.tool()
    async def liquidate_live_algorithm(project_id: int) -> Dict[str, Any]:
        """
        Liquidate all positions in a live algorithm.

        Args:
            project_id: ID of the project with the live algorithm to liquidate

        Returns:
            Dictionary containing liquidation 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}

            # Make API request
            response = await auth.make_authenticated_request(
                endpoint="live/update/liquidate", method="POST", json=request_data
            )

            # Parse response
            if response.status_code == 200:
                data = response.json()

                if data.get("success", False):
                    return {
                        "status": "success",
                        "project_id": project_id,
                        "message": f"Successfully liquidated live algorithm for project {project_id}",
                    }
                else:
                    # API returned success=false
                    errors = data.get("errors", ["Unknown error"])
                    return {
                        "status": "error",
                        "error": "Live algorithm liquidation failed",
                        "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 liquidate live algorithm: {str(e)}",
                "project_id": project_id,
            }

    @mcp.tool()
    async def stop_live_algorithm(project_id: int) -> Dict[str, Any]:
        """
        Stop a live algorithm.

        Args:
            project_id: ID of the project with the live algorithm to stop

        Returns:
            Dictionary containing stop 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}

            # Make API request
            response = await auth.make_authenticated_request(
                endpoint="live/update/stop", method="POST", json=request_data
            )

            # Parse response
            if response.status_code == 200:
                data = response.json()

                if data.get("success", False):
                    return {
                        "status": "success",
                        "project_id": project_id,
                        "message": f"Successfully stopped live algorithm for project {project_id}",
                    }
                else:
                    # API returned success=false
                    errors = data.get("errors", ["Unknown error"])
                    return {
                        "status": "error",
                        "error": "Live algorithm stop failed",
                        "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 stop live algorithm: {str(e)}",
                "project_id": project_id,
            }

    @mcp.tool()
    async def list_live_algorithms(
        status: Optional[str] = None,
        start: Optional[int] = None,
        end: Optional[int] = None,
    ) -> Dict[str, Any]:
        """
        List live algorithms with optional filters.

        Args:
            status: Optional status filter (e.g., "Running", "Stopped")
            start: Optional start time (Unix timestamp)
            end: Optional end time (Unix timestamp)

        Returns:
            Dictionary containing list of live algorithms
        """
        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 = {}
            if status:
                request_data["status"] = status
            if start is not None:
                request_data["start"] = start
            if end is not None:
                request_data["end"] = end

            # Make API request
            response = await auth.make_authenticated_request(
                endpoint="live/list", method="POST", json=request_data
            )

            # Parse response
            if response.status_code == 200:
                data = response.json()

                if data.get("success", False):
                    live_algorithms = data.get("live", [])
                    
                    return {
                        "status": "success",
                        "live_algorithms": live_algorithms,
                        "total_count": len(live_algorithms),
                        "filters": {
                            "status": status,
                            "start": start,
                            "end": end,
                        },
                        "message": f"Successfully retrieved {len(live_algorithms)} live algorithms",
                    }
                else:
                    # API returned success=false
                    errors = data.get("errors", ["Unknown error"])
                    return {
                        "status": "error",
                        "error": "Failed to list live algorithms",
                        "details": errors,
                    }

            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 list live algorithms: {str(e)}",
            }

    @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,
            }

    @mcp.tool()
    async def read_live_chart(
        project_id: int,
        name: str,
        count: int = 100,
        start: Optional[int] = None,
        end: Optional[int] = None,
    ) -> Dict[str, Any]:
        """
        Read chart data from a live algorithm.

        Args:
            project_id: Project ID of the live algorithm
            name: Name of the chart to retrieve (e.g., "Strategy Equity")
            count: Number of data points to request (default: 100)
            start: Optional UTC start timestamp in seconds
            end: Optional UTC end timestamp in seconds

        Returns:
            Dictionary containing live algorithm chart data
        """
        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,
                "name": name,
                "count": count,
            }

            # Add optional timestamp parameters
            if start is not None:
                request_data["start"] = start
            if end is not None:
                request_data["end"] = end

            # Make API request
            response = await auth.make_authenticated_request(
                endpoint="live/chart/read", method="POST", json=request_data
            )

            # Parse response
            if response.status_code == 200:
                data = response.json()

                if data.get("success", False):
                    # Check if chart is still loading
                    if "progress" in data and "status" in data:
                        progress = data.get("progress", 0)
                        status = data.get("status", "loading")
                        return {
                            "status": "loading",
                            "project_id": project_id,
                            "chart_name": name,
                            "progress": progress,
                            "chart_status": status,
                            "message": f"Chart '{name}' is loading... ({progress * 100:.1f}% complete)",
                        }

                    # Chart is ready
                    elif "chart" in data:
                        chart = data.get("chart")
                        return {
                            "status": "success",
                            "project_id": project_id,
                            "chart_name": name,
                            "chart": chart,
                            "count": count,
                            "start": start,
                            "end": end,
                            "message": f"Successfully retrieved chart '{name}' from live algorithm {project_id}",
                        }

                    else:
                        return {
                            "status": "error",
                            "error": "Unexpected response format - no chart or progress data found",
                        }
                else:
                    # API returned success=false
                    errors = data.get("errors", ["Unknown error"])
                    return {
                        "status": "error",
                        "error": "Failed to read live algorithm chart",
                        "details": errors,
                        "project_id": project_id,
                        "chart_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 live algorithm chart: {str(e)}",
                "project_id": project_id,
                "chart_name": name,
            }

    @mcp.tool()
    async def read_live_portfolio(project_id: int) -> Dict[str, Any]:
        """
        Read portfolio state from a live algorithm.

        Args:
            project_id: Project ID of the live algorithm

        Returns:
            Dictionary containing live algorithm portfolio state
        """
        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}

            # Make API request
            response = await auth.make_authenticated_request(
                endpoint="live/portfolio/read", method="POST", json=request_data
            )

            # Parse response
            if response.status_code == 200:
                data = response.json()

                if data.get("success", False):
                    portfolio = data.get("portfolio", {})
                    
                    return {
                        "status": "success",
                        "project_id": project_id,
                        "portfolio": portfolio,
                        "message": f"Successfully retrieved portfolio state from live algorithm {project_id}",
                    }
                else:
                    # API returned success=false
                    errors = data.get("errors", ["Unknown error"])
                    return {
                        "status": "error",
                        "error": "Failed to read live algorithm portfolio",
                        "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 read live algorithm portfolio: {str(e)}",
                "project_id": project_id,
            }

    @mcp.tool()
    async def read_live_orders(
        project_id: int, start: int = 0, end: int = 100
    ) -> Dict[str, Any]:
        """
        Read orders from a live algorithm.

        Args:
            project_id: Project ID of the live algorithm
            start: Starting index of orders to fetch (default: 0)
            end: Last index of orders to fetch (default: 100, max range: 100)

        Returns:
            Dictionary containing live algorithm orders data
        """
        auth = get_auth_instance()
        if auth is None:
            return {
                "status": "error",
                "error": "QuantConnect authentication not configured. Use configure_auth() first.",
            }

        # Validate range
        if end - start > 100:
            return {
                "status": "error",
                "error": "Range too large: end - start must be less than or equal to 100",
            }

        if start < 0 or end < 0:
            return {
                "status": "error",
                "error": "Start and end indices must be non-negative",
            }

        if start >= end:
            return {
                "status": "error",
                "error": "Start index must be less than end index",
            }

        try:
            # Prepare request data
            request_data = {
                "projectId": project_id,
                "start": start,
                "end": end,
            }

            # Make API request
            response = await auth.make_authenticated_request(
                endpoint="live/orders/read", method="POST", json=request_data
            )

            # Parse response
            if response.status_code == 200:
                data = response.json()

                if data.get("success", False):
                    orders = data.get("orders", {})
                    length = data.get("length", 0)

                    return {
                        "status": "success",
                        "project_id": project_id,
                        "start": start,
                        "end": end,
                        "orders": orders,
                        "length": length,
                        "message": f"Successfully retrieved {length} orders from live algorithm {project_id} (range: {start}-{end})",
                    }
                else:
                    # API returned success=false
                    errors = data.get("errors", ["Unknown error"])
                    return {
                        "status": "error",
                        "error": "Failed to read live algorithm orders",
                        "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 read live algorithm orders: {str(e)}",
                "project_id": project_id,
                "start": start,
                "end": end,
            }

    @mcp.tool()
    async def read_live_insights(
        project_id: int, start: int = 0, end: int = 100
    ) -> Dict[str, Any]:
        """
        Read insights from a live algorithm.

        Args:
            project_id: Project ID of the live algorithm
            start: Starting index of insights to fetch (default: 0)
            end: Last index of insights to fetch (default: 100, max range: 100)

        Returns:
            Dictionary containing live algorithm insights data
        """
        auth = get_auth_instance()
        if auth is None:
            return {
                "status": "error",
                "error": "QuantConnect authentication not configured. Use configure_auth() first.",
            }

        # Validate range
        if end - start > 100:
            return {
                "status": "error",
                "error": "Range too large: end - start must be less than or equal to 100",
            }

        if start < 0 or end < 0:
            return {
                "status": "error",
                "error": "Start and end indices must be non-negative",
            }

        if start >= end:
            return {
                "status": "error",
                "error": "Start index must be less than end index",
            }

        try:
            # Prepare request data
            request_data = {
                "projectId": project_id,
                "start": start,
                "end": end,
            }

            # Make API request
            response = await auth.make_authenticated_request(
                endpoint="live/read/insights", method="POST", json=request_data
            )

            # Parse response
            if response.status_code == 200:
                data = response.json()

                if data.get("success", False):
                    insights = data.get("insights", [])
                    length = data.get("length", 0)

                    return {
                        "status": "success",
                        "project_id": project_id,
                        "start": start,
                        "end": end,
                        "insights": insights,
                        "length": length,
                        "message": f"Successfully retrieved {length} insights from live algorithm {project_id} (range: {start}-{end})",
                    }
                else:
                    # API returned success=false
                    errors = data.get("errors", ["Unknown error"])
                    return {
                        "status": "error",
                        "error": "Failed to read live algorithm insights",
                        "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 read live algorithm insights: {str(e)}",
                "project_id": project_id,
                "start": start,
                "end": end,
            }

Tool Definition Quality

B3.2/5.0

Behavior2/5

Does the description disclose side effects, auth requirements, rate limits, or destructive behavior?

No annotations are provided, so the description carries full burden. It states this is a read operation, but doesn't disclose behavioral traits like authentication requirements, rate limits, error conditions, or what 'live' implies (e.g., real-time data vs. cached). The description is minimal and lacks operational context.

Agents need to know what a tool does to the world before calling it. Descriptions should go beyond structured annotations to explain consequences.

Conciseness5/5

Is the description appropriately sized, front-loaded, and free of redundancy?

The description is efficiently structured with a clear purpose statement followed by Args and Returns sections. Every sentence earns its place, and it's front-loaded with the core functionality. No wasted words or redundancy.

Shorter descriptions cost fewer tokens and are easier for agents to parse. Every sentence should earn its place.

Completeness3/5

Given the tool's complexity, does the description cover enough for an agent to succeed on first attempt?

Given the tool has an output schema (returns a dictionary), the description doesn't need to detail return values. However, with no annotations and minimal behavioral disclosure, it's incomplete for a live system tool. The description covers basics but lacks context on permissions, errors, or real-time implications.

Complex tools with many parameters or behaviors need more documentation. Simple tools need less. This dimension scales expectations accordingly.

Parameters4/5

Does the description clarify parameter syntax, constraints, interactions, or defaults beyond what the schema provides?

Schema description coverage is 0%, but the description compensates by explaining the single parameter 'project_id' as 'Project ID of the live algorithm', adding meaningful context beyond the schema's title 'Project Id'. With only one parameter, this is sufficient to understand its purpose, though format or validation details are missing.

Input schemas describe structure but not intent. Descriptions should explain non-obvious parameter relationships and valid value ranges.

Purpose4/5

Does the description clearly state what the tool does and how it differs from similar tools?

The description clearly states the verb 'Read' and resource 'portfolio state from a live algorithm', making the purpose unambiguous. It distinguishes this from other read operations like read_live_algorithm or read_live_orders by specifying it retrieves portfolio state specifically. However, it doesn't explicitly differentiate from all sibling tools beyond the resource scope.

Agents choose between tools based on descriptions. A clear purpose with a specific verb and resource helps agents select the right tool.

Usage Guidelines2/5

Does the description explain when to use this tool, when not to, or what alternatives exist?

No guidance is provided on when to use this tool versus alternatives. The description doesn't mention prerequisites (e.g., requires a running live algorithm), nor does it contrast with similar tools like read_live_algorithm or read_account. Usage context is implied but not stated.

Agents often have multiple tools that could apply. Explicit usage guidance like "use X instead of Y when Z" prevents misuse.

Install Server

Other Tools

Latest Blog Posts

Lightport: Open-Sourcing Glama's AI Gateway
By punkpeye on April 27, 2026.
open source
OpenAI
Tool Definition Quality Score (TDQS)
By punkpeye on April 3, 2026.
mcp
The Hackers Who Tracked My Sleep Cycle
By punkpeye on March 26, 2026.
security

MCP directory API

We provide all the information about MCP servers via our MCP API.

curl -X GET 'https://glama.ai/api/mcp/v1/servers/taylorwilsdon/quantconnect-mcp'

If you have feedback or need assistance with the MCP directory API, please join our Discord server