Jenkins MCP

Instructions

Args:
    job_name: Name of the job to build
    parameters: Optional build parameters as a dictionary (e.g. {"param1": "value1"})

Returns:
    Dictionary containing build information including the build number

Implementation Reference

• jenkins_mcp/server.py:177-251 (handler)

  The complete implementation of the 'trigger_build' tool handler. This function is decorated with @mcp.tool() for registration, includes input validation, docstring describing the schema, verifies the job exists, triggers the build via POST request with CSRF handling, parses the response, and returns build information.

  @mcp.tool()
  def trigger_build(
      ctx: Context, job_name: str, parameters: Optional[dict] = None
  ) -> dict:
      """Trigger a Jenkins build
  
      Args:
          job_name: Name of the job to build
          parameters: Optional build parameters as a dictionary (e.g. {"param1": "value1"})
  
      Returns:
          Dictionary containing build information including the build number
      """
      if not isinstance(job_name, str):
          raise ValueError(f"job_name must be a string, got {type(job_name)}")
      if parameters is not None and not isinstance(parameters, dict):
          raise ValueError(
              f"parameters must be a dictionary or None, got {type(parameters)}"
          )
  
      jenkins_ctx = ctx.request_context.lifespan_context
      client = jenkins_ctx.client
  
      # First verify the job exists
      try:
          job_info = client.get_job_info(job_name)
          if not job_info:
              raise ValueError(f"Job {job_name} not found")
      except Exception as e:
          raise ValueError(f"Error checking job {job_name}: {str(e)}")
  
      # Then try to trigger the build
      try:
          # Get the next build number before triggering
          next_build_number = job_info["nextBuildNumber"]
  
          # Determine the endpoint based on whether parameters are provided
          endpoint = (
              f"job/{job_name}/buildWithParameters"
              if parameters
              else f"job/{job_name}/build"
          )
  
          # Make request with proper CSRF protection
          response = make_jenkins_request(
              jenkins_ctx, "POST", endpoint, params=parameters if parameters else None
          )
  
          if response.status_code not in (200, 201):
              raise ValueError(
                  f"Failed to trigger build: HTTP {response.status_code}, {response.text}"
              )
  
          queue_id = None
          location = response.headers.get("Location")
          if location:
              # Extract queue ID from Location header (e.g., .../queue/item/12345/)
              queue_parts = location.rstrip("/").split("/")
              if queue_parts and queue_parts[-2] == "item":
                  try:
                      queue_id = int(queue_parts[-1])
                  except ValueError:
                      pass
  
          return {
              "status": "triggered",
              "job_name": job_name,
              "queue_id": queue_id,
              "build_number": next_build_number,
              "job_url": job_info["url"],
              "build_url": f"{job_info['url']}{next_build_number}/",
          }
      except Exception as e:
          raise ValueError(f"Error triggering build for {job_name}: {str(e)}")

• jenkins_mcp/server.py:177-177 (registration)

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

  @mcp.tool()

• jenkins_mcp/server.py:62-123 (helper)

  Helper function make_jenkins_request used by trigger_build to make authenticated HTTP requests to Jenkins with automatic CSRF crumb handling and retry logic.

  def make_jenkins_request(
      ctx: JenkinsContext,
      method: str,
      path: str,
      params: Optional[Dict[str, Any]] = None,
      data: Optional[Dict[str, Any]] = None,
      retry_on_auth_failure: bool = True,
  ) -> requests.Response:
      """
      Make a request to Jenkins with proper CSRF protection
  
      Args:
          ctx: Jenkins context with session and auth information
          method: HTTP method (GET, POST, etc.)
          path: Path relative to Jenkins base URL
          params: Query parameters (for GET requests)
          data: Form data (for POST requests)
          retry_on_auth_failure: Whether to retry with a fresh crumb on 403 errors
  
      Returns:
          Response object from the request
      """
      url = urljoin(ctx.jenkins_url, path)
      headers = {}
  
      # Add crumb to headers if available
      if ctx.crumb_data:
          headers.update(ctx.crumb_data)
  
      try:
          response = ctx.session.request(
              method,
              url,
              auth=(ctx.username, ctx.password),
              headers=headers,
              params=params,
              data=data,
          )
  
          # If we get a 403 and it mentions the crumb, try to refresh the crumb and retry
          if (
              response.status_code == 403
              and retry_on_auth_failure
              and ("No valid crumb" in response.text or "Invalid crumb" in response.text)
          ):
              logging.info("Crumb expired, refreshing and retrying request")
              # Get a fresh crumb
              ctx.crumb_data = get_jenkins_crumb(
                  ctx.session, ctx.jenkins_url, ctx.username, ctx.password
              )
              if ctx.crumb_data:
                  # Retry without the retry_on_auth_failure flag to prevent infinite loops
                  return make_jenkins_request(
                      ctx, method, path, params, data, retry_on_auth_failure=False
                  )
  
          return response
      except Exception as e:
          logging.error(f"Error making Jenkins request: {str(e)}")
          raise

• jenkins_mcp/server.py:22-60 (helper)

  Helper function get_jenkins_crumb used to fetch CSRF protection tokens for Jenkins requests, called by make_jenkins_request and lifespan.

  def get_jenkins_crumb(
      session: requests.Session, jenkins_url: str, username: str, password: str
  ) -> Optional[Dict[str, str]]:
      """
      Get a CSRF crumb from Jenkins using the provided session
  
      Args:
          session: The requests Session object to use
          jenkins_url: Base URL of the Jenkins server
          username: Jenkins username
          password: Jenkins password or API token
  
      Returns:
          Dictionary with the crumb field name and value or None if unsuccessful
      """
      try:
          crumb_url = urljoin(jenkins_url, "crumbIssuer/api/json")
  
          response = session.get(crumb_url, auth=(username, password))
          if response.status_code != 200:
              logging.warning(f"Failed to get Jenkins crumb: HTTP {response.status_code}")
              return None
  
          crumb_data = response.json()
          if (
              not crumb_data
              or "crumbRequestField" not in crumb_data
              or "crumb" not in crumb_data
          ):
              logging.warning(f"Invalid crumb response format: {response.text}")
              return None
  
          # Create the crumb header data
          crumb_header = {crumb_data["crumbRequestField"]: crumb_data["crumb"]}
          logging.info(f"Got Jenkins crumb: {crumb_data['crumbRequestField']}=<masked>")
          return crumb_header
      except Exception as e:
          logging.error(f"Error getting Jenkins crumb: {str(e)}")
          return None