Stream

Instructions

Implementation Reference

• src/stream_mcp/tools/invoices.py:30-80 (handler)

  The main handler function create_invoice that creates a ZATCA-compliant invoice in Stream. It accepts customer_id, items (list of dicts with product_id and quantity), scheduled_on, description, currency, notify_consumer, coupons, and various payment method acceptance flags. The function validates items, creates a payment methods object, builds the request body, calls the Stream API via client.post(), and handles errors.

  async def create_invoice(
      customer_id: str,
      items: list[dict],
      scheduled_on: str | None = None,
      description: str | None = None,
      currency: str = "SAR",
      notify_consumer: bool = True,
      coupons: list[str] | None = None,
      accept_mada: bool = True,
      accept_visa: bool = True,
      accept_mastercard: bool = True,
      accept_amex: bool = False,
      accept_bank_transfer: bool = False,
      accept_installment: bool = False,
      ctx: Context = None,  # type: ignore[assignment]
  ) -> dict[str, Any]:
      """Create a ZATCA-compliant invoice in Stream.
  
      *items* is a list of line-item dicts, each with:
        - product_id  (str, required)
        - quantity    (int > 0, required)
  
      *scheduled_on* is the ISO-8601 date-time for when the invoice is due/sent.
      Set *notify_consumer* to True to send the invoice to the customer.
      """
      parsed_items = [InvoiceItemCreateDto(**item) for item in items]
      if scheduled_on is None:
          scheduled_on = datetime.now(timezone.utc).isoformat()
      payment_methods = InvoicePaymentMethodDto(
          mada=accept_mada,
          visa=accept_visa,
          mastercard=accept_mastercard,
          amex=accept_amex,
          bank_transfer=accept_bank_transfer,
          installment=accept_installment,
      )
      body = CreateInvoiceRequest(
          organization_consumer_id=customer_id,
          items=parsed_items,
          payment_methods=payment_methods,
          scheduled_on=scheduled_on,
          notify_consumer=notify_consumer,
          description=description,
          coupons=coupons,
          currency=currency,
      )
      client = await get_client(ctx)
      try:
          return await client.post(_BASE, body.model_dump(exclude_none=True))
      except StreamAPIError as exc:
          return _err(exc)

• src/stream_mcp/models/invoices.py:28-41 (schema)

  Pydantic BaseModel defining the request structure for creating a new invoice. Includes fields: organization_consumer_id (required), items (list of InvoiceItemCreateDto, required), payment_methods (InvoicePaymentMethodDto), scheduled_on (ISO-8601 date-time, required), notify_consumer (boolean), description (string up to 500 chars), coupons (list of strings), and currency (ISO-4217 code, defaults to 'SAR').

  class CreateInvoiceRequest(BaseModel):
      """Request body for creating a new invoice."""
  
      organization_consumer_id: str = Field(..., description="Customer (consumer) ID to invoice.")
      items: list[InvoiceItemCreateDto] = Field(..., min_length=1, description="Line-items on the invoice.")
      payment_methods: InvoicePaymentMethodDto = Field(
          default_factory=InvoicePaymentMethodDto,
          description="Payment methods accepted for this invoice.",
      )
      scheduled_on: str = Field(..., description="ISO-8601 date-time when the invoice should be sent/due.")
      notify_consumer: bool = Field(default=True, description="Send the invoice to the customer upon creation.")
      description: str | None = Field(default=None, max_length=500, description="Invoice memo / description.")
      coupons: list[str] | None = Field(default=None, description="Coupon IDs to apply to the whole invoice.")
      currency: str = Field(default="SAR", description="ISO-4217 currency code.")

• src/stream_mcp/models/invoices.py:8-13 (schema)

  Pydantic BaseModel for a single line-item on an invoice. Requires product_id (string) and quantity (integer > 0). Also accepts an optional coupons field for coupon IDs to apply specifically to this item.

  class InvoiceItemCreateDto(BaseModel):
      """A single line-item on an invoice."""
  
      product_id: str = Field(..., description="ID of the product.")
      quantity: int = Field(..., gt=0, description="Quantity (> 0).")
      coupons: list[str] | None = Field(default=None, description="Coupon IDs to apply to this item.")

• src/stream_mcp/models/invoices.py:16-25 (schema)

  Pydantic BaseModel defining payment methods accepted for an invoice. Contains boolean fields for: mada, visa, mastercard, amex, bank_transfer, installment, and qurrah. All default to False except mada, visa, and mastercard which default to True.

  class InvoicePaymentMethodDto(BaseModel):
      """Payment methods accepted for an invoice."""
  
      mada: bool = Field(default=True, description="Accept Mada payments.")
      visa: bool = Field(default=True, description="Accept Visa payments.")
      mastercard: bool = Field(default=True, description="Accept Mastercard payments.")
      amex: bool = Field(default=False, description="Accept Amex payments.")
      bank_transfer: bool = Field(default=False, description="Accept bank transfer.")
      installment: bool = Field(default=False, description="Accept installment payments.")
      qurrah: bool = Field(default=False, description="Accept Qurrah payments.")

• src/stream_mcp/tools/invoices.py:26-80 (registration)

  The register function that registers all invoice tools on the FastMCP instance. This function contains the @mcp.tool decorator on create_invoice, which is where the tool is registered with the MCP server. The function also defines helper _err for error handling.

  def register(mcp: FastMCP) -> None:
      """Register all invoice tools on *mcp*."""
  
      @mcp.tool
      async def create_invoice(
          customer_id: str,
          items: list[dict],
          scheduled_on: str | None = None,
          description: str | None = None,
          currency: str = "SAR",
          notify_consumer: bool = True,
          coupons: list[str] | None = None,
          accept_mada: bool = True,
          accept_visa: bool = True,
          accept_mastercard: bool = True,
          accept_amex: bool = False,
          accept_bank_transfer: bool = False,
          accept_installment: bool = False,
          ctx: Context = None,  # type: ignore[assignment]
      ) -> dict[str, Any]:
          """Create a ZATCA-compliant invoice in Stream.
  
          *items* is a list of line-item dicts, each with:
            - product_id  (str, required)
            - quantity    (int > 0, required)
  
          *scheduled_on* is the ISO-8601 date-time for when the invoice is due/sent.
          Set *notify_consumer* to True to send the invoice to the customer.
          """
          parsed_items = [InvoiceItemCreateDto(**item) for item in items]
          if scheduled_on is None:
              scheduled_on = datetime.now(timezone.utc).isoformat()
          payment_methods = InvoicePaymentMethodDto(
              mada=accept_mada,
              visa=accept_visa,
              mastercard=accept_mastercard,
              amex=accept_amex,
              bank_transfer=accept_bank_transfer,
              installment=accept_installment,
          )
          body = CreateInvoiceRequest(
              organization_consumer_id=customer_id,
              items=parsed_items,
              payment_methods=payment_methods,
              scheduled_on=scheduled_on,
              notify_consumer=notify_consumer,
              description=description,
              coupons=coupons,
              currency=currency,
          )
          client = await get_client(ctx)
          try:
              return await client.post(_BASE, body.model_dump(exclude_none=True))
          except StreamAPIError as exc:
              return _err(exc)