flatten
Close all open trading positions to manage risk and exit the market. Returns a summary of closed positions for portfolio review.
Instructions
Immediately closes all open positions.
Returns:
Summary of closed positionsInput Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- tools/execution.py:163-186 (handler)The main handler function for the 'flatten' MCP tool. It closes all open positions by calling the broker's close_all_positions method and returns a formatted summary of the action or an error message.
def flatten() -> str: """ Immediately closes all open positions. Returns: Summary of closed positions """ if broker is None: return "ERROR: Alpaca broker not initialized." try: result = broker.close_all_positions() if result['closed_count'] == 0: return "No positions to flatten." msg = [f"✅ Flattened {result['closed_count']} positions:"] for pos in result['positions_closed']: msg.append(f" - {pos['symbol']}: {pos['qty']} shares ({pos['status']})") return "\n".join(msg) except Exception as e: logger.error(f"Flatten failed: {e}") return f"ERROR: Failed to flatten positions - {str(e)}" - server.py:375-378 (registration)Registration of the flatten tool as part of the Execution category tools in the MCP server using the register_tools function, which applies the @mcp.tool() decorator.
register_tools( [place_order, cancel_order, get_positions, flatten, get_order_history], "Execution" ) - tools/alpaca_broker.py:277-301 (helper)The helper method in the AlpacaBroker class that performs the actual closing of all positions via the Alpaca trading client API. This is called by the flatten tool handler.
def close_all_positions(self) -> Dict[str, Any]: """ Close all open positions (emergency flatten). Returns: Dict with closed position count and details """ try: closed = self.trading_client.close_all_positions(cancel_orders=True) logger.info(f"Closed {len(closed)} positions") return { "closed_count": len(closed), "positions_closed": [ { "symbol": pos.symbol, "qty": float(pos.qty) if pos.qty else 0, "status": pos.status.value } for pos in closed ] } except Exception as e: logger.error(f"Failed to close all positions: {e}") raise