Stream

Instructions

Implementation Reference

• src/stream_mcp/tools/customers.py:120-133 (handler)

  The delete_customer handler function - decorated with @mcp.tool, takes customer_id and ctx parameters, performs a soft-delete by calling client.delete() on the /api/v2/consumers/{customer_id} endpoint, and handles StreamAPIError exceptions.

  @mcp.tool
  async def delete_customer(
      customer_id: str,
      ctx: Context = None,  # type: ignore[assignment]
  ) -> dict[str, Any]:
      """Soft-delete a customer by ID.
  
      The customer record is archived but not permanently removed.
      """
      client = await get_client(ctx)
      try:
          return await client.delete(f"{_BASE}/{customer_id}")
      except StreamAPIError as exc:
          return _err(exc)

• src/stream_mcp/tools/customers.py:24-133 (registration)

  The register() function that registers all customer tools including delete_customer with the FastMCP instance using the @mcp.tool decorator.

  def register(mcp: FastMCP) -> None:
      """Register all customer tools on *mcp*."""
  
      @mcp.tool
      async def create_customer(
          name: str,
          phone_number: str | None = None,
          email: str | None = None,
          external_id: str | None = None,
          iban: str | None = None,
          alias: str | None = None,
          comment: str | None = None,
          preferred_language: str | None = None,
          communication_methods: list[str] | None = None,
          ctx: Context = None,  # type: ignore[assignment]
      ) -> dict[str, Any]:
          """Create a new customer in Stream.
  
          Provide at least a *name*. Optionally include *phone_number*, *email*,
          *external_id*, *iban*, *alias*, *comment*, *preferred_language* (EN/AR),
          and *communication_methods* (WHATSAPP, EMAIL, SMS).
          """
          body = CreateCustomerRequest(
              name=name, phone_number=phone_number, email=email,
              external_id=external_id, iban=iban, alias=alias,
              comment=comment, preferred_language=preferred_language,
              communication_methods=communication_methods,
          )
          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_customers(
          page: int = 1,
          limit: int = 20,
          ctx: Context = None,  # type: ignore[assignment]
      ) -> dict[str, Any]:
          """List / search customers with pagination.
  
          Returns a paginated list of customers.
          """
          params: dict[str, Any] = {"page": page, "limit": limit}
          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_customer(
          customer_id: str,
          ctx: Context = None,  # type: ignore[assignment]
      ) -> dict[str, Any]:
          """Get a single customer record by ID."""
          client = await get_client(ctx)
          try:
              return await client.get(f"{_BASE}/{customer_id}")
          except StreamAPIError as exc:
              return _err(exc)
  
      @mcp.tool
      async def update_customer(
          customer_id: str,
          name: str | None = None,
          phone_number: str | None = None,
          email: str | None = None,
          external_id: str | None = None,
          iban: str | None = None,
          alias: str | None = None,
          comment: str | None = None,
          preferred_language: str | None = None,
          communication_methods: list[str] | None = None,
          ctx: Context = None,  # type: ignore[assignment]
      ) -> dict[str, Any]:
          """Update fields on an existing customer.
  
          Only the fields you provide will be changed; others remain untouched.
          """
          body = UpdateCustomerRequest(
              name=name, phone_number=phone_number, email=email,
              external_id=external_id, iban=iban, alias=alias,
              comment=comment, preferred_language=preferred_language,
              communication_methods=communication_methods,
          )
          client = await get_client(ctx)
          try:
              return await client.put(
                  f"{_BASE}/{customer_id}",
                  body.model_dump(exclude_none=True),
              )
          except StreamAPIError as exc:
              return _err(exc)
  
      @mcp.tool
      async def delete_customer(
          customer_id: str,
          ctx: Context = None,  # type: ignore[assignment]
      ) -> dict[str, Any]:
          """Soft-delete a customer by ID.
  
          The customer record is archived but not permanently removed.
          """
          client = await get_client(ctx)
          try:
              return await client.delete(f"{_BASE}/{customer_id}")
          except StreamAPIError as exc:
              return _err(exc)

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

  The _err helper function that formats StreamAPIError exceptions into a consistent error response dict 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/helpers.py:31-73 (helper)

  The get_client async function that retrieves a StreamClient for the current request, supporting both local mode (shared client from lifespan context) and remote mode (per-user client with caching).

  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)

  The register_all_tools function that imports all tool modules and calls their register() functions, including customers.register(mcp) which registers the delete_customer tool.

  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)