Stream

Instructions

Implementation Reference

• src/stream_mcp/tools/coupons.py:28-52 (handler)

  The create_coupon handler function decorated with @mcp.tool. It accepts parameters (name, discount_value, is_percentage, currency, is_active), creates a CreateCouponRequest body, gets an API client, and makes a POST request to create the coupon.

  async def create_coupon(
      name: str,
      discount_value: float,
      is_percentage: bool = False,
      currency: str | None = None,
      is_active: bool = True,
      ctx: Context = None,  # type: ignore[assignment]
  ) -> dict[str, Any]:
      """Create a new discount coupon on Stream.
  
      Set *is_percentage* to True for percentage discount, False for fixed amount.
      For fixed coupons, *currency* is required (e.g. SAR, USD).
      """
      body = CreateCouponRequest(
          name=name,
          discount_value=discount_value,
          is_percentage=is_percentage,
          currency=currency,
          is_active=is_active,
      )
      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/coupons.py:8-18 (schema)

  Pydantic BaseModel for validating create_coupon request parameters. Defines fields for name, discount_value, currency, is_percentage, and is_active with proper constraints and descriptions.

  class CreateCouponRequest(BaseModel):
      """Request body for creating a new coupon."""
  
      name: str = Field(..., min_length=1, max_length=80, description="Coupon display name.")
      discount_value: float = Field(..., ge=0, description="Discount value (amount or percentage).")
      currency: str | None = Field(
          default=None,
          description="ISO-4217 currency code. Required when is_percentage is false. Must be null when is_percentage is true.",
      )
      is_percentage: bool = Field(default=False, description="True for percentage discount, false for fixed amount.")
      is_active: bool = Field(default=True, description="Whether the coupon is active.")

• src/stream_mcp/tools/coupons.py:24-52 (registration)

  The register function that defines and registers the create_coupon tool using the @mcp.tool decorator. This function is called during MCP server initialization to register all coupon-related tools.

  def register(mcp: FastMCP) -> None:
      """Register all coupon tools on *mcp*."""
  
      @mcp.tool
      async def create_coupon(
          name: str,
          discount_value: float,
          is_percentage: bool = False,
          currency: str | None = None,
          is_active: bool = True,
          ctx: Context = None,  # type: ignore[assignment]
      ) -> dict[str, Any]:
          """Create a new discount coupon on Stream.
  
          Set *is_percentage* to True for percentage discount, False for fixed amount.
          For fixed coupons, *currency* is required (e.g. SAR, USD).
          """
          body = CreateCouponRequest(
              name=name,
              discount_value=discount_value,
              is_percentage=is_percentage,
              currency=currency,
              is_active=is_active,
          )
          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/tools/__init__.py:11-30 (registration)

  Main registration orchestrator that imports the coupons module and calls coupons.register(mcp) to register all coupon tools including create_coupon.

  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)

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

  The get_client helper function used by create_coupon to obtain a StreamClient instance. It handles both local (stdio) and remote (SSE/http) modes by retrieving the client from lifespan context or creating a per-user client from the 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]