mcp-server-bitbucket

Instructions

Args:
    repo_slug: Repository slug
    limit: Maximum number of results (default: 10)

Returns:
    List of recent pipeline runs

Implementation Reference

• src/server.py:405-422 (handler)

  MCP tool handler for 'list_pipelines'. Calls BitbucketClient.list_pipelines, formats output using PipelineSummary models, and handles errors.

  @mcp.tool()
  @handle_bitbucket_error
  @formatted
  def list_pipelines(repo_slug: str, limit: int = 10) -> dict:
      """List recent pipeline runs for a repository.
  
      Args:
          repo_slug: Repository slug
          limit: Maximum number of results (default: 10)
  
      Returns:
          List of recent pipeline runs
      """
      client = get_client()
      pipelines = client.list_pipelines(repo_slug, limit=validate_limit(limit))
      return {
          "pipelines": [PipelineSummary.from_api(p).model_dump() for p in pipelines],
      }

• src/bitbucket_client.py:561-579 (helper)

  Core BitbucketClient method implementing the API call to list pipelines via paginated list endpoint.

  def list_pipelines(
      self,
      repo_slug: str,
      limit: int = 10,
  ) -> list[dict[str, Any]]:
      """List recent pipeline runs.
  
      Args:
          repo_slug: Repository slug
          limit: Maximum results to return
  
      Returns:
          List of pipeline info dicts
      """
      return self._paginated_list(
          self._repo_path(repo_slug, "pipelines") + "/",
          limit=limit,
          sort="-created_on",
      )

• src/models.py:258-284 (schema)

  Pydantic model PipelineSummary used for input/output schema validation and transformation in list_pipelines tool response.

  class PipelineSummary(BaseModel):
      """Pipeline info for list responses."""
  
      uuid: str
      build_number: Optional[int] = None
      state: Optional[str] = None
      result: Optional[str] = None
      branch: Optional[str] = None
      created: Optional[str] = None
  
      @field_validator("created", mode="before")
      @classmethod
      def truncate_ts(cls, v: Any) -> Optional[str]:
          return truncate_timestamp(v)
  
      @classmethod
      def from_api(cls, data: dict) -> "PipelineSummary":
          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"),
          )