claude-workspace/jenkins_mcp_server.py

import json, requests
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types
import os, base64

server = Server("jenkins-mcp")
JENKINS_URL = os.environ["JENKINS_URL"].rstrip("/")
JENKINS_USER = os.environ["JENKINS_USER"]
JENKINS_TOKEN = os.environ["JENKINS_TOKEN"]

def auth():
    creds = base64.b64encode(f"{JENKINS_USER}:{JENKINS_TOKEN}".encode()).decode()
    return {"Authorization": f"Basic {creds}"}

def get_crumb():
    """Fetch Jenkins CSRF crumb required for POST/DELETE requests."""
    r = requests.get(
        f"{JENKINS_URL}/crumbIssuer/api/json",
        headers=auth()
    )
    if r.status_code == 200:
        data = r.json()
        return {data["crumbRequestField"]: data["crumb"]}
    return {}

@server.list_tools()
async def list_tools():
    return [
        types.Tool(name="get_jobs", description="Jenkins 잡 목록 조회",
            inputSchema={"type": "object", "properties": {}}),
        types.Tool(name="get_build_status", description="빌드 상태 확인",
            inputSchema={"type": "object", "required": ["job_name"],
                "properties": {
                    "job_name": {"type": "string"},
                    "build_number": {"type": "string", "default": "lastBuild"}
                }}),
        types.Tool(name="trigger_build", description="빌드 트리거",
            inputSchema={"type": "object", "required": ["job_name"],
                "properties": {
                    "job_name": {"type": "string"},
                    "parameters": {"type": "object"}
                }}),
        types.Tool(name="get_build_log", description="빌드 로그 조회",
            inputSchema={"type": "object", "required": ["job_name"],
                "properties": {
                    "job_name": {"type": "string"},
                    "build_number": {"type": "string", "default": "lastBuild"},
                    "lines": {"type": "integer", "default": 100}
                }}),
        types.Tool(name="create_job", description="새 Jenkins 잡 생성 (XML config 필요)",
            inputSchema={"type": "object", "required": ["job_name", "config_xml"],
                "properties": {
                    "job_name": {"type": "string", "description": "생성할 잡 이름"},
                    "config_xml": {"type": "string", "description": "Jenkins job config.xml 내용"}
                }}),
        types.Tool(name="edit_job", description="기존 Jenkins 잡 설정 수정 (XML config 필요)",
            inputSchema={"type": "object", "required": ["job_name", "config_xml"],
                "properties": {
                    "job_name": {"type": "string", "description": "수정할 잡 이름"},
                    "config_xml": {"type": "string", "description": "새 Jenkins job config.xml 내용"}
                }}),
        types.Tool(name="get_job_config", description="Jenkins 잡의 현재 config.xml 조회",
            inputSchema={"type": "object", "required": ["job_name"],
                "properties": {
                    "job_name": {"type": "string"}
                }}),
        types.Tool(name="delete_job", description="Jenkins 잡 삭제",
            inputSchema={"type": "object", "required": ["job_name"],
                "properties": {
                    "job_name": {"type": "string", "description": "삭제할 잡 이름"}
                }}),
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "get_jobs":
        r = requests.get(
            f"{JENKINS_URL}/api/json?tree=jobs[name,color,url]",
            headers=auth()
        )
        jobs = r.json().get("jobs", [])
        return [types.TextContent(type="text", text=json.dumps(jobs, ensure_ascii=False))]

    elif name == "get_build_status":
        job = arguments["job_name"]
        num = arguments.get("build_number", "lastBuild")
        r = requests.get(f"{JENKINS_URL}/job/{job}/{num}/api/json", headers=auth())
        return [types.TextContent(type="text", text=json.dumps(r.json(), ensure_ascii=False))]

    elif name == "trigger_build":
        job = arguments["job_name"]
        params = arguments.get("parameters", {})
        headers = {**auth(), **get_crumb()}
        if params:
            r = requests.post(f"{JENKINS_URL}/job/{job}/buildWithParameters",
                headers=headers, params=params)
        else:
            r = requests.post(f"{JENKINS_URL}/job/{job}/build", headers=headers)
        return [types.TextContent(type="text", text=f"Status: {r.status_code}")]

    elif name == "get_build_log":
        job = arguments["job_name"]
        num = arguments.get("build_number", "lastBuild")
        lines = arguments.get("lines", 100)
        r = requests.get(
            f"{JENKINS_URL}/job/{job}/{num}/logText/progressiveText",
            headers=auth()
        )
        log_lines = r.text.split("\n")[-lines:]
        return [types.TextContent(type="text", text="\n".join(log_lines))]

    elif name == "get_job_config":
        job = arguments["job_name"]
        r = requests.get(f"{JENKINS_URL}/job/{job}/config.xml", headers=auth())
        if r.status_code == 200:
            return [types.TextContent(type="text", text=r.text)]
        return [types.TextContent(type="text", text=f"Error {r.status_code}: {r.text}")]

    elif name == "create_job":
        job = arguments["job_name"]
        config_xml = arguments["config_xml"]
        headers = {**auth(), **get_crumb(), "Content-Type": "application/xml"}
        r = requests.post(
            f"{JENKINS_URL}/createItem?name={job}",
            headers=headers,
            data=config_xml.encode("utf-8")
        )
        if r.status_code in (200, 201):
            return [types.TextContent(type="text", text=f"Job '{job}' created successfully.")]
        return [types.TextContent(type="text", text=f"Error {r.status_code}: {r.text}")]

    elif name == "edit_job":
        job = arguments["job_name"]
        config_xml = arguments["config_xml"]
        headers = {**auth(), **get_crumb(), "Content-Type": "application/xml"}
        r = requests.post(
            f"{JENKINS_URL}/job/{job}/config.xml",
            headers=headers,
            data=config_xml.encode("utf-8")
        )
        if r.status_code == 200:
            return [types.TextContent(type="text", text=f"Job '{job}' updated successfully.")]
        return [types.TextContent(type="text", text=f"Error {r.status_code}: {r.text}")]

    elif name == "delete_job":
        job = arguments["job_name"]
        headers = {**auth(), **get_crumb()}
        r = requests.post(f"{JENKINS_URL}/job/{job}/doDelete", headers=headers)
        if r.status_code in (200, 302):
            return [types.TextContent(type="text", text=f"Job '{job}' deleted successfully.")]
        return [types.TextContent(type="text", text=f"Error {r.status_code}: {r.text}")]

    return [types.TextContent(type="text", text=f"Unknown tool: {name}")]

if __name__ == "__main__":
    import asyncio
    async def main():
        async with stdio_server() as (read_stream, write_stream):
            await server.run(read_stream, write_stream, server.create_initialization_options())
    asyncio.run(main())