Stream

Instructions

Implementation Reference

• src/stream_mcp/tools/products.py:29-64 (handler)

  The create_product tool handler function that creates a new product in Stream. It accepts product parameters (name, type, price, currency, etc.), creates a CreateProductRequest body, and posts to the API via the StreamClient.

  async def create_product(
      name: str,
      type: Literal["ONE_OFF", "RECURRING", "METERED"] = "ONE_OFF",
      price: float = 1.0,
      currency: str = "SAR",
      description: str | None = None,
      is_price_inclusive_of_vat: bool = True,
      is_price_exempt_from_vat: bool = False,
      recurring_interval: str | None = None,
      recurring_interval_count: int = 1,
      ctx: Context = None,  # type: ignore[assignment]
  ) -> dict[str, Any]:
      """Create a new product or service in Stream.
  
      *type* is ``ONE_OFF``, ``RECURRING``, or ``METERED``.
      For recurring products, specify *recurring_interval* (WEEK, MONTH, SEMESTER, YEAR).
      """
      prices = [ProductPriceInlineCreate(
          currency=currency,
          amount=price,
          is_price_inclusive_of_vat=is_price_inclusive_of_vat,
          is_price_exempt_from_vat=is_price_exempt_from_vat,
      )]
      body = CreateProductRequest(
          name=name,
          type=type,
          description=description,
          prices=prices,
          recurring_interval=recurring_interval,
          recurring_interval_count=recurring_interval_count,
      )
      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/products.py:19-34 (schema)

  CreateProductRequest Pydantic model defining the schema for creating products with validation for name, type, description, prices, recurring intervals, etc.

  class CreateProductRequest(BaseModel):
      """Request body for creating a new product."""
  
      name: str = Field(..., min_length=1, max_length=160, description="Product display name.")
      type: Literal["ONE_OFF", "RECURRING", "METERED"] = Field(
          ...,
          description="Product billing type: ONE_OFF, RECURRING, or METERED.",
      )
      description: str | None = Field(default=None, max_length=500, description="Product description.")
      prices: list[ProductPriceInlineCreate] | None = Field(default=None, description="List of prices for different currencies.")
      is_one_time: bool = Field(default=False, description="Whether the product is one-time use.")
      recurring_interval: str | None = Field(
          default=None,
          description="Billing interval for recurring products: WEEK, MONTH, SEMESTER, YEAR.",
      )
      recurring_interval_count: int = Field(default=1, ge=1, description="Number of intervals between billings.")

• src/stream_mcp/models/products.py:10-16 (schema)

  ProductPriceInlineCreate Pydantic model defining the price structure (currency, amount, VAT flags) used within CreateProductRequest.

  class ProductPriceInlineCreate(BaseModel):
      """Inline price definition for product creation."""
  
      currency: str = Field(default="SAR", description="ISO-4217 currency code (SAR, USD, EUR, etc.).")
      amount: float = Field(..., ge=1, description="Price amount (≥ 1).")
      is_price_inclusive_of_vat: bool = Field(default=True, description="Whether price includes VAT.")
      is_price_exempt_from_vat: bool = Field(default=False, description="Whether price is VAT-exempt.")

• src/stream_mcp/tools/products.py:25-64 (registration)

  The register function that registers the create_product tool with FastMCP using the @mcp.tool decorator.

  def register(mcp: FastMCP) -> None:
      """Register all product tools on *mcp*."""
  
      @mcp.tool
      async def create_product(
          name: str,
          type: Literal["ONE_OFF", "RECURRING", "METERED"] = "ONE_OFF",
          price: float = 1.0,
          currency: str = "SAR",
          description: str | None = None,
          is_price_inclusive_of_vat: bool = True,
          is_price_exempt_from_vat: bool = False,
          recurring_interval: str | None = None,
          recurring_interval_count: int = 1,
          ctx: Context = None,  # type: ignore[assignment]
      ) -> dict[str, Any]:
          """Create a new product or service in Stream.
  
          *type* is ``ONE_OFF``, ``RECURRING``, or ``METERED``.
          For recurring products, specify *recurring_interval* (WEEK, MONTH, SEMESTER, YEAR).
          """
          prices = [ProductPriceInlineCreate(
              currency=currency,
              amount=price,
              is_price_inclusive_of_vat=is_price_inclusive_of_vat,
              is_price_exempt_from_vat=is_price_exempt_from_vat,
          )]
          body = CreateProductRequest(
              name=name,
              type=type,
              description=description,
              prices=prices,
              recurring_interval=recurring_interval,
              recurring_interval_count=recurring_interval_count,
          )
          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 create_product to obtain a StreamClient for making API requests, supporting both local (lifespan) and remote (Bearer token) modes.

  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]