mcp-server-bitbucket

Instructions

Args:
    repo_slug: Repository slug
    branch: Branch to run pipeline on (default: main)
    variables: Custom pipeline variables as key-value pairs (optional)

Returns:
    Pipeline run info with uuid and state

Implementation Reference

• src/server.py:353-382 (handler)

  MCP tool handler function decorated with @mcp.tool(). Executes the trigger_pipeline tool by calling the BitbucketClient method and formatting the response.

  @mcp.tool()
  @handle_bitbucket_error
  @formatted
  def trigger_pipeline(
      repo_slug: str,
      branch: str = "main",
      variables: Optional[dict] = None,
  ) -> dict:
      """Trigger a pipeline run on a repository.
  
      Args:
          repo_slug: Repository slug
          branch: Branch to run pipeline on (default: main)
          variables: Custom pipeline variables as key-value pairs (optional)
  
      Returns:
          Pipeline run info with uuid and state
      """
      client = get_client()
      result = client.trigger_pipeline(
          repo_slug=repo_slug,
          branch=branch,
          variables=variables,
      )
      return {
          "uuid": result.get("uuid"),
          "build_number": result.get("build_number"),
          "state": result.get("state", {}).get("name"),
      }

• src/bitbucket_client.py:508-541 (helper)

  BitbucketClient helper method that makes the actual Bitbucket API POST request to trigger the pipeline.

  def trigger_pipeline(
      self,
      repo_slug: str,
      branch: str = "main",
      variables: Optional[dict[str, str]] = None,
  ) -> dict[str, Any]:
      """Trigger a pipeline run.
  
      Args:
          repo_slug: Repository slug
          branch: Branch to run pipeline on (default: main)
          variables: Custom pipeline variables
  
      Returns:
          Pipeline run info including 'uuid', 'state'
      """
      payload = {
          "target": {
              "ref_type": "branch",
              "type": "pipeline_ref_target",
              "ref_name": branch,
          }
      }
      if variables:
          payload["variables"] = [
              {"key": k, "value": v} for k, v in variables.items()
          ]
  
      result = self._request(
          "POST",
          self._repo_path(repo_slug, "pipelines") + "/",
          json=payload,
      )
      return self._require_result(result, "trigger pipeline on", branch)