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())