build_user_context

Instructions

Implementation Reference

• src/pagerduty_mcp_server/users.py:16-68 (handler)

  Core implementation of build_user_context. Fetches the current PagerDuty user via _show_current_user(), then builds a context dict containing user_id, name, email, team_ids (from teams.fetch_team_ids), service_ids (from services.fetch_service_ids), and escalation_policy_ids (from escalation_policies.fetch_escalation_policy_ids). Includes error handling via utils.handle_api_error.

  async def build_user_context() -> Dict[str, Any]:
      """Validate and build the current user's context. Exposed as MCP server tool.
  
      See the "Standard Response Format" section in `tools.md` for the complete standard response structure.
  
      Returns:
          See the "Standard Response Format" section in `tools.md` for the complete standard response structure.
          The response will contain the current user's context in the format defined in the "build_user_context" section of `tools.md`.
  
      Raises:
          See the "Error Handling" section in `tools.md` for common error scenarios.
      """
      try:
          user = await _show_current_user()
          if not user:
              raise ValueError("Failed to get current user data")
  
          context = {
              "user_id": str(user.get("id", "")).strip(),
              "name": user.get("name", ""),
              "email": user.get("email", ""),
              "team_ids": [],
              "service_ids": [],
              "escalation_policy_ids": [],
          }
  
          if not context["user_id"]:
              raise ValueError("Invalid user data: missing or empty user ID")
  
          team_ids = teams.fetch_team_ids(user=user)
          context["team_ids"] = [
              str(tid).strip() for tid in team_ids if tid and str(tid).strip()
          ]
  
          if context["team_ids"]:
              service_ids = await services.fetch_service_ids(team_ids=context["team_ids"])
              context["service_ids"] = [
                  str(sid).strip() for sid in service_ids if sid and str(sid).strip()
              ]
  
          escalation_policy_ids = await escalation_policies.fetch_escalation_policy_ids(
              user_id=context["user_id"]
          )
          context["escalation_policy_ids"] = [
              str(epid).strip()
              for epid in escalation_policy_ids
              if epid and str(epid).strip()
          ]
  
          return context
  
      except Exception as e:
          utils.handle_api_error(e)

• src/pagerduty_mcp_server/server.py:585-602 (registration)

  MCP tool registration for build_user_context. Decorated with @mcp.tool() and @tool_error_boundary, delegates directly to users.build_user_context().

  @mcp.tool()
  @tool_error_boundary
  async def build_user_context() -> Dict[str, Any]:
      """Validate and build the current user's context into a dictionary with the following format:
          {
              "user_id": str,
              "team_ids": List[str],
              "service_ids": List[str],
              "escalation_policy_ids": List[str]
          }
      The MCP server tools use this user context to filter the following resources:
          - Escalation policies
          - Incidents
          - Oncalls
          - Services
          - Users
      """
      return await users.build_user_context()

• src/pagerduty_mcp_server/users.py:33-40 (schema)

  Return type definition/schema of build_user_context. Returns a dict with keys: user_id (str), name (str), email (str), team_ids (List[str]), service_ids (List[str]), escalation_policy_ids (List[str]).

  context = {
      "user_id": str(user.get("id", "")).strip(),
      "name": user.get("name", ""),
      "email": user.get("email", ""),
      "team_ids": [],
      "service_ids": [],
      "escalation_policy_ids": [],
  }

• src/pagerduty_mcp_server/users.py:169-193 (helper)

  Private helper _show_current_user() called by build_user_context. Calls PagerDuty API GET /users/me to retrieve the current user's profile, validates the response has an 'id', and returns the parsed user dict.

  async def _show_current_user() -> Dict[str, Any]:
      """Get the current user's PagerDuty profile including their teams, contact methods, and notification rules.
  
      Returns:
          See the "Standard Response Format" section in `tools.md` for the complete standard response structure.
          The response will contain a single user with detailed profile information including teams, contact methods, and notification rules.
  
      Raises:
          See the "Error Handling" section in `tools.md` for common error scenarios.
      """
      pd_client = create_client()
      try:
          result = await safe_execute_async(
              lambda: pd_client.jget(USERS_URL + "/me"), "fetch current user"
          )
          response = result["user"]
          user = {}
          if response:
              model = User.model_validate(response)
              user = model.to_clean_dict()
          if not user or "id" not in user or not user["id"]:
              raise ValueError("Invalid user object: missing ID")
          return user
      except Exception as e:
          utils.handle_api_error(e)