Stream

Instructions

Implementation Reference

• src/stream_mcp/tools/products.py:93-103 (handler)

  The get_product tool handler - takes a product_id parameter, uses the Stream API client to fetch the product from /api/v2/products/{product_id}, and handles API errors gracefully.

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

• src/stream_mcp/models/products.py:49-60 (schema)

  ProductResponse schema - defines the expected response structure when fetching a product from the Stream API, including id, name, description, price, currency, type, and created_at fields.

  class ProductResponse(BaseModel):
      """Subset of fields returned by the Stream API for a product."""
  
      id: str
      name: str | None = None
      description: str | None = None
      price: float | None = None
      currency: str | None = None
      type: str | None = None
      created_at: str | None = None
  
      model_config = {"extra": "allow"}

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

  The register function that registers all product tools including get_product on the FastMCP instance. This function is called by register_all_tools during server initialization.

  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)
  
      @mcp.tool
      async def list_products(
          page: int = 1,
          limit: int = 20,
          type: str | None = None,
          active: bool | None = None,
          currency: str | None = None,
          ctx: Context = None,  # type: ignore[assignment]
      ) -> dict[str, Any]:
          """List products with optional filters.
  
          *type* can be ``ONE_OFF``, ``RECURRING``, or ``METERED``.
          *active* filters by active/inactive status.
          """
          params: dict[str, Any] = {"page": page, "limit": limit}
          if type:
              params["type"] = type
          if active is not None:
              params["active"] = active
          if currency:
              params["currency"] = currency
          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_product(
          product_id: str,
          ctx: Context = None,  # type: ignore[assignment]
      ) -> dict[str, Any]:
          """Get a single product by ID."""
          client = await get_client(ctx)
          try:
              return await client.get(f"{_BASE}/{product_id}")
          except StreamAPIError as exc:
              return _err(exc)
  
      @mcp.tool
      async def update_product(
          product_id: str,
          name: str | None = None,
          description: str | None = None,
          is_active: bool | None = None,
          ctx: Context = None,  # type: ignore[assignment]
      ) -> dict[str, Any]:
          """Update an existing product's name, description, or active status.
  
          Only the fields you provide will be changed.
          """
          body = UpdateProductRequest(
              name=name, description=description, is_active=is_active,
          )
          client = await get_client(ctx)
          try:
              return await client.put(
                  f"{_BASE}/{product_id}",
                  body.model_dump(exclude_none=True),
              )
          except StreamAPIError as exc:
              return _err(exc)
  
      @mcp.tool
      async def archive_product(
          product_id: str,
          ctx: Context = None,  # type: ignore[assignment]
      ) -> dict[str, Any]:
          """Archive a product so it can no longer be sold.
  
          This is a soft-delete; the product record is retained for history.
          """
          client = await get_client(ctx)
          try:
              return await client.delete(f"{_BASE}/{product_id}")
          except StreamAPIError as exc:
              return _err(exc)

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

  get_client helper function - returns a StreamClient for the current request context, supporting 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]

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

  Central registration point - register_all_tools imports all tool modules including products and calls their register functions, which in turn registers get_product on the FastMCP instance.

  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)