Stream

Instructions

Implementation Reference

• src/stream_mcp/tools/payment_links.py:88-98 (handler)

  The get_payment_link tool handler - retrieves a single payment link by ID using the Stream API client. Accepts payment_link_id parameter and returns the API response or error details.

  @mcp.tool
  async def get_payment_link(
      payment_link_id: str,
      ctx: Context = None,  # type: ignore[assignment]
  ) -> dict[str, Any]:
      """Retrieve a single payment link by its ID."""
      client = await get_client(ctx)
      try:
          return await client.get(f"{_BASE}/{payment_link_id}")
      except StreamAPIError as exc:
          return _err(exc)

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

  The register function that registers all payment-link tools including get_payment_link. This is called by register_all_tools to expose the tool to the MCP server.

  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)
  
      @mcp.tool
      async def list_payment_links(
          page: int = 1,
          limit: int = 20,
          statuses: list[str] | None = None,
          from_date: str | None = None,
          to_date: str | None = None,
          ctx: Context = None,  # type: ignore[assignment]
      ) -> dict[str, Any]:
          """List all payment links with optional filters.
  
          *statuses* can include: INACTIVE, ACTIVE, COMPLETED.
          """
          params: dict[str, Any] = {"page": page, "limit": limit}
          if statuses:
              params["statuses"] = statuses
          if from_date:
              params["from_date"] = from_date
          if to_date:
              params["to_date"] = to_date
          client = await get_client(ctx)
          try:
              return await client.get(_BASE, params=params)
          except StreamAPIError as exc:
              return _err(exc)
  
      @mcp.tool
      async def get_payment_link(
          payment_link_id: str,
          ctx: Context = None,  # type: ignore[assignment]
      ) -> dict[str, Any]:
          """Retrieve a single payment link by its ID."""
          client = await get_client(ctx)
          try:
              return await client.get(f"{_BASE}/{payment_link_id}")
          except StreamAPIError as exc:
              return _err(exc)
  
      @mcp.tool
      async def deactivate_payment_link(
          payment_link_id: str,
          deactivate_message: str | None = None,
          ctx: Context = None,  # type: ignore[assignment]
      ) -> dict[str, Any]:
          """Deactivate (archive) a payment link so it can no longer be used."""
          body = UpdatePaymentLinkStatusRequest(
              status="INACTIVE",
              deactivate_message=deactivate_message,
          )
          client = await get_client(ctx)
          try:
              return await client.patch(
                  f"{_BASE}/{payment_link_id}/status",
                  body.model_dump(exclude_none=True),
              )
          except StreamAPIError as exc:
              return _err(exc)

• src/stream_mcp/models/payment_links.py:38-48 (schema)

  PaymentLinkResponse schema defining the structure of data returned when retrieving a payment link (id, url, status, name, description, created_at, with extra fields allowed).

  class PaymentLinkResponse(BaseModel):
      """Subset of fields returned by the Stream API for a payment link."""
  
      id: str
      url: str | None = None
      status: str | None = None
      name: str | None = None
      description: str | None = None
      created_at: str | None = None
  
      model_config = {"extra": "allow"}

• src/stream_mcp/tools/payment_links.py:20-21 (helper)

  Error helper function that formats StreamAPIError exceptions into a consistent dict structure with error, code, message, and details fields.

  def _err(exc: StreamAPIError) -> dict[str, Any]:
      return {"error": True, "code": exc.status_code, "message": str(exc), "details": exc.body}

• src/stream_mcp/tools/__init__.py:11-30 (registration)

  register_all_tools function that imports and calls payment_links.register(mcp) to make all payment link tools available to the MCP server.

  def register_all_tools(mcp: FastMCP) -> None:
      """Import every tool / resource module and call its ``register(mcp)``."""
      from stream_mcp.tools import (
          coupons,
          customers,
          docs,
          invoices,
          payment_links,
          payments,
          products,
      )
  
      payment_links.register(mcp)
      customers.register(mcp)
      products.register(mcp)
      payments.register(mcp)
      coupons.register(mcp)
      invoices.register(mcp)
      docs.register(mcp)