create_live_algorithm
Deploy compiled trading algorithms to live brokerage connections for automated execution with real market data.
Instructions
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
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| project_id | Yes | ||
| compile_id | Yes | ||
| node_id | Yes | ||
| brokerage_id | Yes | ||
| brokerage_config | Yes | ||
| data_providers | No | ||
| version_id | No | -1 | |
| parameters | No | ||
| notifications | No |
Implementation Reference
- 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):