Stream

Instructions

Implementation Reference

• src/stream_mcp/tools/payment_links.py:28-60 (handler)

  The main create_payment_link tool handler function. It accepts parameters (name, items, description, currency, valid_until, max_number_of_payments, organization_consumer_id), creates a CreatePaymentLinkRequest, gets the API client, and makes a POST request to /api/v2/payment_links. Handles StreamAPIError exceptions.

  async def create_payment_link(
      name: str,
      items: list[dict],
      description: str | None = None,
      currency: str = "SAR",
      valid_until: str | None = None,
      max_number_of_payments: int | None = None,
      organization_consumer_id: str | None = None,
      ctx: Context = None,  # type: ignore[assignment]
  ) -> dict[str, Any]:
      """Create a new payment / checkout link on Stream.
  
      *items* is a list of objects, each containing:
        - product_id (str, required)
        - quantity   (int ≥ 1, optional, default 1)
        - coupons    (list[str], optional)
  
      You **cannot** mix one-time and recurring products in the same link.
      """
      body = CreatePaymentLinkRequest(
          name=name,
          items=items,  # type: ignore[arg-type]
          description=description,
          currency=currency,
          valid_until=valid_until,
          max_number_of_payments=max_number_of_payments,
          organization_consumer_id=organization_consumer_id,
      )
      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/payment_links.py:16-28 (schema)

  Pydantic BaseModel for CreatePaymentLinkRequest with field validation. Defines the structure for creating payment links including name, items, description, currency, coupons, max_number_of_payments, valid_until, success_redirect_url, failure_redirect_url, and organization_consumer_id.

  class CreatePaymentLinkRequest(BaseModel):
      """Request body for creating a new payment link."""
  
      name: str = Field(..., description="Display name for the payment link.")
      items: list[PaymentLinkItem] = Field(..., min_length=1, description="Line-items (product + quantity + coupons). Cannot mix one-time and recurring products.")
      description: str | None = Field(default=None, description="Optional description shown on the checkout page.")
      currency: str = Field(default="SAR", description="ISO-4217 currency code.")
      coupons: list[str] | None = Field(default=None, description="Coupon IDs to apply.")
      max_number_of_payments: int | None = Field(default=None, description="Maximum number of payments allowed.")
      valid_until: str | None = Field(default=None, description="ISO-8601 expiry timestamp. Null = never expires.")
      success_redirect_url: str | None = Field(default=None, description="URL to redirect to after successful payment.")
      failure_redirect_url: str | None = Field(default=None, description="URL to redirect to after failed payment.")
      organization_consumer_id: str | None = Field(default=None, description="Pre-assign a customer to this link.")

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

  PaymentLinkItem nested schema defining individual line items with product_id (required), quantity (≥1), and optional coupons list.

  class PaymentLinkItem(BaseModel):
      """A single line-item inside a payment link."""
  
      product_id: str = Field(..., description="ID of the product to include.")
      quantity: int = Field(ge=1, description="Number of units (≥ 1).")
      coupons: list[str] = Field(default_factory=list, description="Coupon IDs to apply to this item.")

• src/stream_mcp/tools/payment_links.py:24-60 (registration)

  The register function that registers create_payment_link as an MCP tool using @mcp.tool decorator. This function is called from the tools/__init__.py module.

  def register(mcp: FastMCP) -> None:
      """Register all payment-link tools on *mcp*."""
  
      @mcp.tool
      async def create_payment_link(
          name: str,
          items: list[dict],
          description: str | None = None,
          currency: str = "SAR",
          valid_until: str | None = None,
          max_number_of_payments: int | None = None,
          organization_consumer_id: str | None = None,
          ctx: Context = None,  # type: ignore[assignment]
      ) -> dict[str, Any]:
          """Create a new payment / checkout link on Stream.
  
          *items* is a list of objects, each containing:
            - product_id (str, required)
            - quantity   (int ≥ 1, optional, default 1)
            - coupons    (list[str], optional)
  
          You **cannot** mix one-time and recurring products in the same link.
          """
          body = CreatePaymentLinkRequest(
              name=name,
              items=items,  # type: ignore[arg-type]
              description=description,
              currency=currency,
              valid_until=valid_until,
              max_number_of_payments=max_number_of_payments,
              organization_consumer_id=organization_consumer_id,
          )
          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/helpers.py:31-73 (helper)

  The get_client helper function used by the handler to obtain a StreamClient instance. Supports both local mode (shared client from lifespan) and remote mode (per-user client from Bearer token).

  async def get_client(ctx: "Context") -> StreamClient:
      """Return a :class:`StreamClient` for the current request.
  
      Resolution order:
  
      1. **Lifespan client** — used in local / stdio mode where a single
         ``STREAM_API_KEY`` is set as an environment variable.
      2. **Per-user client** — used in remote mode where each user passes
         their own API key as a Bearer token and (optionally) a custom
         base URL via the ``X-Stream-Base-URL`` header.
      """
      # ── 1. Local mode: shared client from server lifespan ─────────────
      shared_client = ctx.lifespan_context.get("client")
      if shared_client is not None:
          return shared_client
  
      # ── 2. Remote mode: per-user client from Bearer token ─────────────
      api_key = current_api_key.get()
      if not api_key:
          raise StreamError(
              "No Stream API key found. "
              "In local mode, set the STREAM_API_KEY env var. "
              "In remote mode, pass your key as a Bearer token in the Authorization header."
          )
  
      base_url = current_base_url.get() or settings.stream_base_url
      cache_key = f"{api_key}::{base_url}"
  
      if cache_key not in _client_cache:
          client = StreamClient(
              api_key=api_key,
              base_url=base_url,
              timeout=settings.stream_timeout,
              max_retries=settings.stream_max_retries,
          )
          await client.__aenter__()
          _client_cache[cache_key] = client
          logger.info(
              "Created cached StreamClient for remote user (key=…%s, base=%s)",
              api_key[-4:], base_url,
          )
  
      return _client_cache[cache_key]