Stream

Instructions

Implementation Reference

• src/stream_mcp/tools/customers.py:28-56 (handler)

  The main handler function for create_customer tool. Decorated with @mcp.tool, it accepts customer parameters (name, phone_number, email, etc.), validates them using CreateCustomerRequest model, and makes a POST request to the Stream API to create the customer.

  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)

• src/stream_mcp/models/customers.py:8-19 (schema)

  Pydantic BaseModel schema for input validation of create_customer tool parameters. Defines all fields with descriptions and validation rules.

  class CreateCustomerRequest(BaseModel):
      """Request body for creating a new customer."""
  
      name: str = Field(..., description="Full name of the customer.")
      phone_number: str | None = Field(default=None, description="Customer phone number (E.164 recommended).")
      email: str | None = Field(default=None, description="Customer email address.")
      external_id: str | None = Field(default=None, description="External system ID for the customer.")
      iban: str | None = Field(default=None, max_length=34, description="Customer IBAN (max 34 chars).")
      alias: str | None = Field(default=None, description="Customer alias / nickname.")
      comment: str | None = Field(default=None, description="Internal comment about the customer.")
      preferred_language: str | None = Field(default=None, description="Preferred language (e.g. EN, AR).")
      communication_methods: list[str] | None = Field(default=None, description="Communication methods: WHATSAPP, EMAIL, SMS.")

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

  The register function that contains all customer tool registrations, including the create_customer tool decorated with @mcp.tool.

  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)

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

  Central registration file that imports all tool modules and calls their register functions. The customers.register(mcp) call at line 24 registers the create_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)

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

  Helper function get_client used by the create_customer handler to obtain a StreamClient instance. Supports both local (shared client) and remote (per-user client) 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]