Commerce-MCP

Implementation Reference

• tools/order_collect.py:9-26 (handler)

  The core handler function that executes the commerce_collect_orders tool logic. It initializes a database connection, validates incoming orders against StandardOrder model, saves them to the database, and returns the count and list of all stored orders.

  def collect_orders(
      *,
      orders: list[dict[str, object]] | None = None,
      db_path: Path | str,
  ) -> dict[str, object]:
      database = Database(db_path)
      database.bootstrap()
  
      if orders:
          normalized = [StandardOrder.model_validate(order) for order in orders]
          database.save_orders(normalized)
  
      stored_orders = database.list_orders()
      return {
          "count": len(stored_orders),
          "orders": stored_orders,
      }

• core/models.py:74-88 (schema)

  Pydantic schema defining the StandardOrder model used for input validation in the collect_orders handler. Defines fields like id, channel, channel_order_id, product_id, buyer info, quantity, price, status, and timestamps.

  class StandardOrder(BaseModel):
      id: str = Field(default_factory=generate_order_id)
      channel: str
      channel_order_id: str
      product_id: str
      buyer_name: str
      buyer_phone: str | None = None
      address: str | None = None
      quantity: int = 1
      selling_price: int = 0
      status: str = "pending"
      invoice_number: str | None = None
      ordered_at: str = Field(default_factory=utc_now_iso)
      processed_at: str | None = None

• core/server.py:58-60 (registration)

  Registration point where the commerce_collect_orders tool is registered with the FastMCP app using the @app.tool decorator. This exposes the tool to MCP clients.

  @app.tool(name="commerce_collect_orders")
  def commerce_collect_orders(orders: list[dict[str, object]] | None = None) -> dict[str, object]:
      return collect_orders(orders=orders, db_path=resolved_db_path)