Stream

Instructions

Implementation Reference

• src/stream_mcp/tools/customers.py:75-85 (handler)

  The get_customer tool handler function decorated with @mcp.tool. Takes a customer_id parameter, gets the API client, makes a GET request to /api/v2/consumers/{customer_id}, and handles StreamAPIError exceptions.

  @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)

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

  The register function that defines and registers all customer tools including get_customer on the FastMCP instance.

  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/models/customers.py:36-45 (schema)

  CustomerResponse Pydantic model defining the expected structure of customer data returned by the API (used as schema documentation for the get_customer tool response).

  class CustomerResponse(BaseModel):
      """Subset of fields returned by the Stream API for a customer."""
  
      id: str
      name: str | None = None
      email: str | None = None
      phone: str | None = None
      created_at: str | None = None
  
      model_config = {"extra": "allow"}

• src/stream_mcp/tools/__init__.py:23-24 (registration)

  Calls customers.register(mcp) to register all customer tools including get_customer onto the FastMCP instance.

  payment_links.register(mcp)
  customers.register(mcp)

• src/stream_mcp/server.py:65-66 (registration)

  Server entry point that calls register_all_tools(mcp) to register all tools including get_customer on the FastMCP instance.

  # Register all tools & resources onto the FastMCP instance
  register_all_tools(mcp)