mcp-server-bitbucket

Instructions

Args:
    repo_slug: Repository slug
    pipeline_uuid: Pipeline UUID (from trigger_pipeline)

Returns:
    Pipeline status including state, duration, and steps

Implementation Reference

• src/server.py:384-402 (handler)

  The MCP tool handler function for 'get_pipeline'. It validates input, calls the BitbucketClient.get_pipeline method, handles not found cases, and returns a formatted PipelineDetail model.

  @mcp.tool()
  @handle_bitbucket_error
  @formatted
  def get_pipeline(repo_slug: str, pipeline_uuid: str) -> dict:
      """Get status of a pipeline run.
  
      Args:
          repo_slug: Repository slug
          pipeline_uuid: Pipeline UUID (from trigger_pipeline)
  
      Returns:
          Pipeline status including state, duration, and steps
      """
      client = get_client()
      result = client.get_pipeline(repo_slug, pipeline_uuid)
      if not result:
          return not_found_response("Pipeline", pipeline_uuid)
  
      return PipelineDetail.from_api(result).model_dump()

• src/bitbucket_client.py:543-559 (helper)

  The BitbucketClient helper method that performs the actual API request to retrieve pipeline details from Bitbucket.

  def get_pipeline(
      self, repo_slug: str, pipeline_uuid: str
  ) -> Optional[dict[str, Any]]:
      """Get pipeline run status.
  
      Args:
          repo_slug: Repository slug
          pipeline_uuid: Pipeline UUID (with or without braces)
  
      Returns:
          Pipeline info or None if not found
      """
      pipeline_uuid = ensure_uuid_braces(pipeline_uuid)
      return self._request(
          "GET",
          self._repo_path(repo_slug, "pipelines", pipeline_uuid),
      )

• src/models.py:287-318 (schema)

  Pydantic model defining the output schema for pipeline details, including from_api factory method to parse Bitbucket API responses.

  class PipelineDetail(BaseModel):
      """Pipeline info for get responses."""
  
      uuid: str
      build_number: Optional[int] = None
      state: Optional[str] = None
      result: Optional[str] = None
      branch: Optional[str] = None
      created: Optional[str] = None
      completed: Optional[str] = None
      duration_s: Optional[int] = None
  
      @field_validator("created", "completed", mode="before")
      @classmethod
      def truncate_ts(cls, v: Any) -> Optional[str]:
          return truncate_timestamp(v)
  
      @classmethod
      def from_api(cls, data: dict) -> "PipelineDetail":
          state_data = data.get("state") or {}
          result_data = state_data.get("result") or {}
          return cls(
              uuid=data.get("uuid", ""),
              build_number=data.get("build_number"),
              state=state_data.get("name"),
              result=result_data.get("name") if result_data else None,
              branch=(data.get("target") or {}).get("ref_name"),
              created=data.get("created_on"),
              completed=data.get("completed_on"),
              duration_s=data.get("duration_in_seconds"),
          )

• src/server.py:384-384 (registration)

  The @mcp.tool() decorator registers the get_pipeline function as an MCP tool.

  @mcp.tool()