jenkins_get_job_config

Instructions

Implementation Reference

• src/jenkins_mcp_server/tools.py:119-122 (handler)

  The actual tool handler function 'jenkins_get_job_config' that calls _get_text with the job path appended with /config.xml

  @mcp.tool()
  def jenkins_get_job_config(job: str | list[str]) -> dict[str, Any]:
      """Read job config.xml. Jenkins may redact secrets without Configure permission."""
      return _run(lambda: _get_text(f"{job_path(job)}/config.xml"))

• src/jenkins_mcp_server/tools.py:55-55 (registration)

  The register_tools function that wraps the handler in @mcp.tool() decorator, registering it with the MCP server

  def register_tools(mcp: FastMCP) -> None:

• src/jenkins_mcp_server/client.py:67-182 (helper)

  The job_path helper function that converts a job name (string or list) to the Jenkins URL path format with 'job' prefix segments

  def job_path(job: str | list[str]) -> str:
      pieces = [piece for piece in job.split("/") if piece] if isinstance(job, str) else job
      if not pieces:
          raise PathValidationError("job must include at least one path segment")
  
      encoded: list[str] = []
      for piece in pieces:
          if not piece or piece in {".", ".."} or "/" in piece:
              raise PathValidationError("job path segments must be non-empty names")
          encoded.extend(["job", quote(piece, safe="")])
      return "/".join(encoded)
  
  
  def safe_segment(value: str, label: str) -> str:
      if not value or value in {".", ".."} or "/" in value:
          raise PathValidationError(f"{label} must be a single Jenkins path segment")
      return quote(value, safe="")
  
  
  class JenkinsClient:
      def __init__(
          self,
          config: JenkinsConfig,
          *,
          transport: httpx.BaseTransport | None = None,
          crumb_manager: CrumbManager | None = None,
      ) -> None:
          self.config = config
          self.crumbs = crumb_manager or CrumbManager()
          auth = None
          if config.user and config.api_token:
              auth = httpx.BasicAuth(config.user, config.api_token)
          self.http = httpx.Client(
              auth=auth,
              verify=config.verify_ssl,
              timeout=config.timeout_seconds,
              follow_redirects=False,
              transport=transport,
          )
  
      @classmethod
      def from_env(cls) -> JenkinsClient:
          return cls(JenkinsConfig.from_env())
  
      def close(self) -> None:
          self.http.close()
  
      def __enter__(self) -> JenkinsClient:
          return self
  
      def __exit__(self, *args: object) -> None:
          self.close()
  
      def _url(self, path: str) -> tuple[str, str]:
          relative = normalize_relative_path(path)
          return self.config.url + relative, relative
  
      def _raise_for_status(self, response: httpx.Response, method: str, path: str) -> None:
          if response.status_code < 400:
              return
          reason = response.reason_phrase or "Jenkins request failed"
          raise JenkinsHTTPError(
              status_code=response.status_code,
              method=method,
              path=path,
              message=reason,
              body=_body_snippet(response),
          )
  
      def request(
          self,
          method: str,
          path: str,
          *,
          params: Mapping[str, Any] | None = None,
          data: Mapping[str, Any] | None = None,
          content: str | bytes | None = None,
          headers: Mapping[str, str] | None = None,
          max_bytes: int | None = None,
      ) -> httpx.Response:
          method = method.upper()
          if method not in {"GET", "POST"}:
              raise PathValidationError("Only GET and POST are supported internally")
          url, relative = self._url(path)
          limit = max_bytes or self.config.max_response_bytes
          request_headers = dict(headers or {})
  
          if method == "POST":
              try:
                  crumb = self.crumbs.get(self.http, self.config.url)
              except httpx.HTTPStatusError:
                  crumb = None
              if crumb is not None:
                  request_headers[crumb.request_field] = crumb.crumb
  
          response = self.http.request(
              method,
              url,
              params=params,
              data=data,
              content=content,
              headers=request_headers,
          )
          if method == "POST" and response.status_code == 403 and _body_snippet(response):
              body = _body_snippet(response, 1000) or ""
              if "crumb" in body.lower():
                  self.crumbs.clear()
                  crumb = self.crumbs.get(self.http, self.config.url)
                  retry_headers = dict(request_headers)
                  if crumb is not None:
                      retry_headers[crumb.request_field] = crumb.crumb
                  response = self.http.request(
                      method,
                      url,
                      params=params,
                      data=data,

• src/jenkins_mcp_server/tools.py:32-34 (helper)

  The _get_text helper that creates a Jenkins client and calls get_text to retrieve raw text content

  def _get_text(path: str) -> str:
      with _client() as client:
          return client.get_text(path)

• src/jenkins_mcp_server/client.py:206-208 (helper)

  The JenkinsClient.get_text method that performs an HTTP GET and returns the raw response text

  def get_text(self, path: str, *, params: Mapping[str, Any] | None = None) -> str:
      response = self.request("GET", path, params=params)
      return response.text