read_live_orders
Retrieve live trading orders from a QuantConnect algorithm to monitor execution status and manage positions in real-time.
Instructions
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
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| project_id | Yes | ||
| start | No | ||
| end | No |
Implementation Reference
- The core handler function for the 'read_live_orders' tool. Decorated with @mcp.tool(), validates inputs, authenticates, makes API request to QuantConnect's live/orders/read endpoint, and returns orders data or error.@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, }
- quantconnect_mcp/src/server.py:78-78 (registration)Registers all live trading tools, including 'read_live_orders', by calling register_live_tools(mcp) during MCP server initialization.register_live_tools(mcp)
- Input schema defined by function parameters (project_id: int required, start/end: int optional defaults) and comprehensive docstring. Output is Dict[str, Any] with status, orders, etc.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 """