create_dca_order

Instructions

Create a DCA order.

Expects a DCAOrderRequestContainer, returns a list of DCAOrderResponse.

Implementation Reference

• armor_crypto_mcp/armor_mcp.py:418-432 (handler)

  MCP tool handler for 'create_dca_order'. This is the primary execution function for the MCP tool, decorated with @mcp.tool() which handles registration. It validates input via type hints, checks authentication, calls the underlying client API, and returns results or errors.

  @mcp.tool()
  async def create_dca_order(dca_order_requests: DCAOrderRequestContainer) -> List[DCAOrderResponse]:
      """
      Create a DCA order.
      
      Expects a DCAOrderRequestContainer, returns a list of DCAOrderResponse.
      """
      if not armor_client:
          return [{"error": "Not logged in"}]
      try:
          result: List[DCAOrderResponse] = await armor_client.create_dca_order(dca_order_requests)
          return result
      except Exception as e:
          return [{"error": str(e)}]

• armor_crypto_mcp/armor_client.py:669-673 (helper)

  Underlying client method in ArmorWalletAPIClient that serializes the DCAOrderRequestContainer to JSON payload and makes a POST API call to the backend endpoint 'transactions/dca-order/create/' to execute the DCA order creation.

  async def create_dca_order(self, data: DCAOrderRequestContainer) -> List[DCAOrderResponse]:
      """Create a DCA order."""
      # payload = [v.model_dump() for v in data.dca_order_requests]
      payload = data.model_dump(exclude_none=True)['dca_order_requests']
      return await self._api_call("POST", "transactions/dca-order/create/", payload)

• armor_crypto_mcp/armor_client.py:469-471 (schema)

  Pydantic container model for input: wraps a list of DCAOrderRequest objects, used for batch operations.

  class DCAOrderRequestContainer(BaseModel):
      dca_order_requests: List[DCAOrderRequest]

• armor_crypto_mcp/armor_client.py:235-249 (schema)

  Pydantic model defining the input schema for a single DCA order request, including wallet, tokens, amount, schedule, duration, execution type, and optional conditional watchers.

  class DCAOrderRequest(BaseModel):
      wallet: str = Field(description="name of the wallet")
      input_token: str = Field(description="public address of the input token. To get the address from a token symbol use `get_token_details`")
      output_token: str = Field(description="public address of the output token. To get the address from a token symbol use `get_token_details`")
      amount: float = Field(description="total amount of input token to invest")
      cron_expression: str = Field(description="cron expression for the DCA worker execution frequency")
      strategy_duration_unit: Literal["MINUTE", "HOUR", "DAY", "WEEK", "MONTH", "YEAR"] = Field(description="unit of the duration of the DCA order")
      strategy_duration: int = Field(description="Total running time of the DCA order given in strategy duration units, should be more than 0")
      execution_type: Literal["MULTIPLE", "SINGLE"] = Field(description="set to SINGLE only if the user is asking for a single scheduled order, MULTIPLE if it is a true DCA")
      token_address_watcher: Optional[str] = Field(description="If the DCA is conditional, public address of the token to watch.")
      watch_field: Optional[Literal["liquidity", "marketCap", "price"]] = Field(description="If the DCA is conditional, field to watch for the condition")
      delta_type: Optional[Literal["INCREASE", "DECREASE", "MOVE", "MOVE_DAILY", "AVERAGE_MOVE"]] = Field(description="If the DCA is conditional, the operator of the watch field in the conditional statement")
      delta_percentage: Optional[float] = Field(description="If the DCA is conditional, percentage of the change to watch for given the delta_type")
      time_zone: Optional[str] = Field(description="user's time zone. Defaults to UTC")

• armor_crypto_mcp/armor_client.py:264-278 (schema)

  Pydantic model defining the output schema for a DCA order response, including order details, status, token info, watchers, and transactions.

  class DCAOrderResponse(BaseModel):
      id: str = Field(description="id of the DCA order")
      amount: float = Field(description="amount of tokens to invest")
      investment_per_cycle: float = Field(description="amount of tokens to invest per cycle")
      cycles_completed: int = Field(description="number of cycles completed")
      total_cycles: int = Field(description="total number of cycles")
      human_readable_expiry: str = Field(description="human readable expiry date of the DCA order")
      status: str = Field(description="status of the DCA order")
      input_token_data: TokenData = Field(description="details of the input token")
      output_token_data: TokenData = Field(description="details of the output token")
      wallet_name: str = Field(description="name of the wallet")
      watchers: List[DCAWatcher] = Field(description="list of watchers for the DCA order")
      dca_transactions: List[dict] = Field(description="list of DCA transactions")  # Can be further typed if structure is known
      created: str = Field(description="Linux timestamp of the creation of the order")