mcp-jenkins

import os
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from fastmcp import Context, FastMCP
from fastmcp.server.dependencies import get_http_request
from loguru import logger
from pydantic import BaseModel, ConfigDict

from mcp_jenkins.jenkins import Jenkins


class LifespanContext(BaseModel):
    model_config = ConfigDict(arbitrary_types_allowed=True)

    jenkins_url: str | None
    jenkins_username: str | None
    jenkins_password: str | None
    jenkins_timeout: int = 5
    jenkins_verify_ssl: bool = True

    read_only: bool = False
    tool_regex: str = ''
    jenkins_session_singleton: bool = True


@asynccontextmanager
async def lifespan(app: FastMCP[LifespanContext]) -> AsyncIterator['LifespanContext']:
    jenkins_url = os.getenv('jenkins_url')
    jenkins_username = os.getenv('jenkins_username')
    jenkins_password = os.getenv('jenkins_password')

    jenkins_timeout = int(os.getenv('jenkins_timeout', '5'))
    jenkins_verify_ssl = os.getenv('jenkins_verify_ssl', 'true').lower() == 'true'
    jenkins_session_singleton = os.getenv('jenkins_session_singleton', 'true').lower() == 'true'

    read_only = os.getenv('read_only', 'false').lower() == 'true'
    tool_regex = os.getenv('tool_regex', '')

    yield LifespanContext(
        jenkins_url=jenkins_url,
        jenkins_username=jenkins_username,
        jenkins_password=jenkins_password,
        jenkins_timeout=jenkins_timeout,
        jenkins_verify_ssl=jenkins_verify_ssl,
        jenkins_session_singleton=jenkins_session_singleton,
        read_only=read_only,
        tool_regex=tool_regex,
    )


def jenkins(ctx: Context) -> Jenkins:
    if ctx.request_context.lifespan_context.jenkins_session_singleton and getattr(ctx.session, 'jenkins', None):
        return ctx.session.jenkins

    jenkins_url = ctx.request_context.lifespan_context.jenkins_url
    jenkins_username = ctx.request_context.lifespan_context.jenkins_username
    jenkins_password = ctx.request_context.lifespan_context.jenkins_password

    jenkins_timeout = ctx.request_context.lifespan_context.jenkins_timeout
    jenkins_verify_ssl = ctx.request_context.lifespan_context.jenkins_verify_ssl

    try:
        requests = get_http_request()

        jenkins_url = getattr(requests.state, 'jenkins_url', jenkins_url)
        jenkins_username = getattr(requests.state, 'jenkins_username', jenkins_username)
        jenkins_password = getattr(requests.state, 'jenkins_password', jenkins_password)

        logger.debug(f'Retrieved Jenkins auth from request state - url: {jenkins_url}, username: {jenkins_username}')
    except RuntimeError as e:
        logger.debug(f'No HTTP request context available, falling back to environment variables: {e}')
    except Exception as e:  # noqa: BLE001
        logger.error(
            f'Unexpected error retrieving Jenkins auth from request, falling back to environment variables: {e}'
        )

    if not all((jenkins_url, jenkins_username, jenkins_password)):
        msg = (
            'Jenkins authentication details are missing. '
            'Please provide them via x-jenkins-* headers '
            'or CLI arguments (--jenkins-url, --jenkins-username, --jenkins-password).'
        )
        raise ValueError(msg)

    logger.info(
        f'Creating Jenkins client with url: '
        f'{jenkins_url}, username: {jenkins_username}, timeout: {jenkins_timeout}, verify_ssl: {jenkins_verify_ssl}'
    )

    ctx.session.jenkins = Jenkins(
        url=jenkins_url,
        username=jenkins_username,
        password=jenkins_password,
        timeout=jenkins_timeout,
        verify_ssl=jenkins_verify_ssl,
    )

    return ctx.session.jenkins