Stream

Instructions

Implementation Reference

• src/stream_mcp/tools/payments.py:58-71 (handler)

  The get_payment tool handler - decorated with @mcp.tool and implements the logic to retrieve a single payment by ID using the Stream API client

  @mcp.tool
  async def get_payment(
      payment_id: str,
      ctx: Context = None,  # type: ignore[assignment]
  ) -> dict[str, Any]:
      """Get details of a single payment by ID.
  
      Returns amount, status, payment method, customer info, and more.
      """
      client = await get_client(ctx)
      try:
          return await client.get(f"{_BASE}/{payment_id}")
      except StreamAPIError as exc:
          return _err(exc)

• src/stream_mcp/tools/payments.py:21-122 (registration)

  The register function that registers all payment tools including get_payment, which is called by register_all_tools during server initialization

  def register(mcp: FastMCP) -> None:
      """Register all payment tools on *mcp*."""
  
      @mcp.tool
      async def list_payments(
          page: int = 1,
          limit: int = 20,
          statuses: list[str] | None = None,
          invoice_id: str | None = None,
          search_term: str | None = None,
          from_date: str | None = None,
          to_date: str | None = None,
          ctx: Context = None,  # type: ignore[assignment]
      ) -> dict[str, Any]:
          """List payments with optional filters.
  
          Filter by *statuses* (PENDING, PROCESSING, SUCCEEDED, FAILED, CANCELED,
          UNDER_REVIEW, EXPIRED, SETTLED, REFUNDED), *invoice_id*, *search_term*,
          or a date range (*from_date* / *to_date* in ISO-8601).
          """
          params: dict[str, Any] = {"page": page, "limit": limit}
          if statuses:
              params["statuses"] = statuses
          if invoice_id:
              params["invoice_id"] = invoice_id
          if search_term:
              params["search_term"] = search_term
          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(
          payment_id: str,
          ctx: Context = None,  # type: ignore[assignment]
      ) -> dict[str, Any]:
          """Get details of a single payment by ID.
  
          Returns amount, status, payment method, customer info, and more.
          """
          client = await get_client(ctx)
          try:
              return await client.get(f"{_BASE}/{payment_id}")
          except StreamAPIError as exc:
              return _err(exc)
  
      @mcp.tool
      async def mark_payment_as_paid(
          payment_id: str,
          payment_method: str = "CASH",
          note: str | None = None,
          ctx: Context = None,  # type: ignore[assignment]
      ) -> dict[str, Any]:
          """Manually mark a payment as paid.
  
          Record a payment received through manual methods.
          *payment_method* must be one of: CASH, BANK_TRANSFER, CARD, or QURRAH.
          """
          body = MarkPaymentPaidRequest(
              payment_method=payment_method,
              note=note,
          )
          client = await get_client(ctx)
          try:
              return await client.post(
                  f"{_BASE}/{payment_id}/mark-paid",
                  body.model_dump(exclude_none=True),
              )
          except StreamAPIError as exc:
              return _err(exc)
  
      @mcp.tool
      async def refund_payment(
          payment_id: str,
          refund_reason: str = "REQUESTED_BY_CUSTOMER",
          refund_note: str | None = None,
          allow_refund_multiple_related_payments: bool = False,
          ctx: Context = None,  # type: ignore[assignment]
      ) -> dict[str, Any]:
          """Issue a refund on a completed payment.
  
          *refund_reason* must be one of: REQUESTED_BY_CUSTOMER, DUPLICATE, FRAUDULENT, OTHER.
          """
          body = RefundPaymentRequest(
              refund_reason=refund_reason,
              refund_note=refund_note,
              allow_refund_multiple_related_payments=allow_refund_multiple_related_payments,
          )
          client = await get_client(ctx)
          try:
              return await client.post(
                  f"{_BASE}/{payment_id}/refund",
                  body.model_dump(exclude_none=True),
              )
          except StreamAPIError as exc:
              return _err(exc)

• src/stream_mcp/models/payments.py:31-42 (schema)

  PaymentResponse Pydantic model defining the schema/structure of payment data returned by the get_payment tool

  class PaymentResponse(BaseModel):
      """Subset of fields returned by the Stream API for a payment."""
  
      id: str
      amount: float | None = None
      currency: str | None = None
      status: str | None = None
      customer_id: str | None = None
      payment_method: str | None = None
      created_at: str | None = None
  
      model_config = {"extra": "allow"}

• src/stream_mcp/helpers.py:31-73 (helper)

  get_client helper function used by the get_payment handler to retrieve the appropriate Stream API client (shared in local mode, per-user in remote mode)

  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]

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

  register_all_tools function that imports and calls payments.register(mcp) to register the get_payment tool during server startup

  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)