mcp_github_jira_server.py•53.6 kB
"""MCP Server for GitHub and Jira Integration"""
import asyncio
import json
import logging
import os
import secrets
import webbrowser
import subprocess
import platform
import sys
from typing import Any, Dict, List, Optional, Union
from urllib.parse import quote, urlencode
from http.server import HTTPServer, BaseHTTPRequestHandler
import threading
import time
import httpx
from mcp.server.fastmcp import FastMCP
from mcp.types import (
Resource,
Tool,
TextContent,
ImageContent,
EmbeddedResource,
LoggingLevel,
Prompt,
PromptArgument,
)
from dotenv import load_dotenv
load_dotenv()
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def load_config():
global GITHUB_TOKEN, JIRA_URL, JIRA_CLIENT_ID, JIRA_CLIENT_SECRET, JIRA_ACCESS_TOKEN, JIRA_REFRESH_TOKEN
GITHUB_TOKEN = os.getenv("GITHUB_TOKEN")
JIRA_URL = os.getenv("JIRA_URL")
JIRA_CLIENT_ID = os.getenv("JIRA_CLIENT_ID")
JIRA_CLIENT_SECRET = os.getenv("JIRA_CLIENT_SECRET")
JIRA_ACCESS_TOKEN = os.getenv("JIRA_ACCESS_TOKEN")
JIRA_REFRESH_TOKEN = os.getenv("JIRA_REFRESH_TOKEN")
if not all([JIRA_URL, JIRA_CLIENT_ID, JIRA_CLIENT_SECRET]):
try:
config_path = "mcp_config.json"
if os.path.exists(config_path):
with open(config_path, 'r') as f:
config = json.load(f)
mcp_server = config.get("mcpServers", {}).get("github-jira", {})
env_vars = mcp_server.get("env", {})
if not JIRA_URL and env_vars.get("JIRA_URL"):
JIRA_URL = env_vars["JIRA_URL"]
if not JIRA_CLIENT_ID and env_vars.get("JIRA_CLIENT_ID"):
JIRA_CLIENT_ID = env_vars["JIRA_CLIENT_ID"]
if not JIRA_CLIENT_SECRET and env_vars.get("JIRA_CLIENT_SECRET"):
JIRA_CLIENT_SECRET = env_vars["JIRA_CLIENT_SECRET"]
if not GITHUB_TOKEN and env_vars.get("GITHUB_TOKEN"):
GITHUB_TOKEN = env_vars["GITHUB_TOKEN"]
logger.info(f"Loaded configuration from {config_path} (fallback)")
else:
logger.warning(f"Configuration file {config_path} not found")
except Exception as e:
logger.error(f"Error loading configuration: {e}")
logger.info(f"Configuration loaded - JIRA_URL: {JIRA_URL}, Client ID: {JIRA_CLIENT_ID[:8] if JIRA_CLIENT_ID else 'None'}...")
load_config()
mcp = FastMCP("github-jira-integration")
github_client = httpx.AsyncClient(
headers={
"Authorization": f"token {GITHUB_TOKEN}",
"Accept": "application/vnd.github.v3+json",
"User-Agent": "MCP-GitHub-Jira-Server/1.0"
} if GITHUB_TOKEN else None
)
jira_client = None
github_user = None
jira_user = None
oauth_server = None
oauth_code = None
async def get_github_user() -> Optional[str]:
global github_user
if not GITHUB_TOKEN or github_user:
return github_user
try:
response = await github_client.get("https://api.github.com/user")
if response.status_code == 200:
github_user = response.json()["login"]
return github_user
except Exception as e:
logger.error(f"Failed to get GitHub user: {e}")
return None
async def refresh_token_if_needed() -> bool:
"""Automatically refresh Jira token if needed and update client"""
global JIRA_ACCESS_TOKEN, JIRA_REFRESH_TOKEN, jira_client
if not JIRA_REFRESH_TOKEN:
return False
if not JIRA_CLIENT_ID or not JIRA_CLIENT_SECRET:
return False
try:
refresh_data = {
"grant_type": "refresh_token",
"client_id": JIRA_CLIENT_ID,
"client_secret": JIRA_CLIENT_SECRET,
"refresh_token": JIRA_REFRESH_TOKEN
}
response = httpx.post(
"https://auth.atlassian.com/oauth/token",
data=refresh_data,
headers={"Content-Type": "application/x-www-form-urlencoded"},
timeout=30.0
)
if response.status_code == 200:
tokens = response.json()
JIRA_ACCESS_TOKEN = tokens["access_token"]
if "refresh_token" in tokens:
JIRA_REFRESH_TOKEN = tokens["refresh_token"]
os.environ["JIRA_REFRESH_TOKEN"] = JIRA_REFRESH_TOKEN
os.environ["JIRA_ACCESS_TOKEN"] = JIRA_ACCESS_TOKEN
update_env_file("JIRA_ACCESS_TOKEN", JIRA_ACCESS_TOKEN)
if "refresh_token" in tokens:
update_env_file("JIRA_REFRESH_TOKEN", JIRA_REFRESH_TOKEN)
jira_client = None
logger.info("[AUTO-REFRESH] Jira access token refreshed successfully!")
return True
else:
logger.warning(f"[AUTO-REFRESH] Token refresh failed: {response.status_code}")
return False
except Exception as e:
logger.warning(f"[AUTO-REFRESH] Token refresh failed with exception: {str(e)}")
return False
async def ensure_jira_client() -> bool:
global jira_client
if jira_client:
return True
if not JIRA_ACCESS_TOKEN:
return False
jira_client = httpx.AsyncClient(
headers={
"Authorization": f"Bearer {JIRA_ACCESS_TOKEN}",
"Accept": "application/json",
"User-Agent": "MCP-GitHub-Jira-Server/1.0"
}
)
return True
async def make_jira_request(method: str, url: str, **kwargs) -> httpx.Response:
"""Make a Jira API request with automatic token refresh on 401"""
if not await ensure_jira_client():
raise Exception("Jira client not available. Please run setup_jira_oauth.")
response = await jira_client.request(method, url, **kwargs)
if response.status_code == 401:
logger.info("[AUTO-REFRESH] Access token expired, attempting refresh...")
if await refresh_token_if_needed():
if await ensure_jira_client():
logger.info("[AUTO-REFRESH] Retrying request with new token...")
response = await jira_client.request(method, url, **kwargs)
if response.status_code != 401:
logger.info("[AUTO-REFRESH] Request successful after token refresh!")
else:
logger.error("[AUTO-REFRESH] Token refresh failed. Manual OAuth required.")
return response
def update_env_file(key: str, value: str) -> bool:
"""Update or add a key-value pair in the .env file."""
try:
env_path = ".env"
lines = []
if os.path.exists(env_path):
with open(env_path, 'r', encoding='utf-8') as f:
lines = f.readlines()
key_found = False
for i, line in enumerate(lines):
if line.strip().startswith(f"{key}="):
lines[i] = f"{key}={value}\n"
key_found = True
break
if not key_found:
if lines and not lines[-1].endswith('\n'):
lines.append('\n')
lines.append(f"{key}={value}\n")
with open(env_path, 'w', encoding='utf-8') as f:
f.writelines(lines)
return True
except Exception as e:
logger.error(f"Error updating .env file: {e}")
return False
def open_browser_with_fallback(url: str) -> bool:
"""Try to open browser with multiple fallback methods."""
try:
if webbrowser.open(url):
return True
except Exception as e:
logger.debug(f"webbrowser.open failed: {e}")
try:
if platform.system() == "Windows":
methods = [
lambda: os.startfile(url),
lambda: subprocess.run(["cmd", "/c", "start", "", url], check=True, shell=False),
lambda: subprocess.run(["powershell", "-Command", f"Start-Process '{url}'"], check=True),
lambda: subprocess.run(["explorer", url], check=True)
]
for i, method in enumerate(methods, 1):
try:
method()
logger.info(f"Successfully opened browser using Windows method {i}")
return True
except Exception as e:
logger.debug(f"Windows method {i} failed: {e}")
continue
elif platform.system() == "Darwin":
subprocess.run(["open", url], check=True)
return True
elif platform.system() == "Linux":
subprocess.run(["xdg-open", url], check=True)
return True
except Exception as e:
logger.debug(f"Platform-specific browser opening failed: {e}")
return False
@mcp.tool("github_read_file")
async def github_read_file(owner: str, repo: str, path: str) -> str:
try:
current_user = await get_github_user()
is_own_repo = current_user and owner.lower() == current_user.lower()
if not is_own_repo:
repo_response = await github_client.get(
f"https://api.github.com/repos/{owner}/{repo}"
)
if repo_response.status_code != 200 or repo_response.json().get("private", False):
raise PermissionError(f"Access denied to private repository {owner}/{repo}")
response = await github_client.get(
f"https://api.github.com/repos/{owner}/{repo}/contents/{quote(path)}"
)
if response.status_code == 200:
content = response.json()
if content.get("type") == "file":
import base64
file_content = base64.b64decode(content["content"]).decode("utf-8")
return f"File content from {owner}/{repo}/{path}:\n\n{file_content}"
else:
return f"Path {path} is not a file"
else:
return f"File {path} not found in {owner}/{repo}"
except Exception as e:
logger.error(f"Error reading GitHub file: {e}")
return f"Error: {str(e)}"
@mcp.tool("github_create_issue")
async def github_create_issue(owner: str, repo: str, title: str, body: str, labels: List[str] = None) -> str:
try:
current_user = await get_github_user()
if not current_user or owner.lower() != current_user.lower():
raise PermissionError(f"Can only create issues in your own repositories. Current user: {current_user}, Target owner: {owner}")
issue_data = {
"title": title,
"body": body
}
if labels:
issue_data["labels"] = labels
response = await github_client.post(
f"https://api.github.com/repos/{owner}/{repo}/issues",
json=issue_data
)
if response.status_code == 201:
issue = response.json()
return f"Successfully created issue #{issue['number']}: {issue['title']}\nURL: {issue['html_url']}"
else:
return f"Failed to create issue: {response.status_code} - {response.text}"
except Exception as e:
logger.error(f"Error creating GitHub issue: {e}")
return f"Error: {str(e)}"
@mcp.tool("github_create_pull_request")
async def github_create_pull_request(owner: str, repo: str, title: str, body: str, head: str, base: str = "main") -> str:
try:
current_user = await get_github_user()
if not current_user or owner.lower() != current_user.lower():
raise PermissionError(f"Can only create pull requests in your own repositories. Current user: {current_user}, Target owner: {owner}")
pr_data = {
"title": title,
"body": body,
"head": head,
"base": base
}
response = await github_client.post(
f"https://api.github.com/repos/{owner}/{repo}/pulls",
json=pr_data
)
if response.status_code == 201:
pr = response.json()
return f"Successfully created pull request #{pr['number']}: {pr['title']}\nURL: {pr['html_url']}"
else:
return f"Failed to create pull request: {response.status_code} - {response.text}"
except Exception as e:
logger.error(f"Error creating GitHub PR: {e}")
return f"Error: {str(e)}"
@mcp.tool("jira_create_issue")
async def jira_create_issue(project_key: str, summary: str, description: str, issue_type: str = "Task") -> str:
try:
resources_response = await make_jira_request(
"GET",
"https://api.atlassian.com/oauth/token/accessible-resources"
)
if resources_response.status_code != 200:
return f"Failed to get accessible resources: {resources_response.status_code}"
resources = resources_response.json()
if not resources:
return "No accessible Jira resources found."
cloud_id = resources[0].get('id')
if not cloud_id:
return "Could not retrieve Cloud ID."
issue_data = {
"fields": {
"project": {"key": project_key},
"summary": summary,
"description": {"type": "doc", "version": 1, "content": [{"type": "paragraph", "content": [{"type": "text", "text": description}]}]},
"issuetype": {"name": issue_type}
}
}
response = await make_jira_request(
"POST",
f"https://api.atlassian.com/ex/jira/{cloud_id}/rest/api/3/issue",
json=issue_data
)
if response.status_code == 201:
issue = response.json()
return f"Successfully created Jira issue {issue['key']}: {summary}\nURL: {JIRA_URL}/browse/{issue['key']}"
else:
return f"Failed to create Jira issue: {response.status_code} - {response.text}"
except Exception as e:
logger.error(f"Error creating Jira issue: {e}")
return f"Error: {str(e)}"
@mcp.tool("jira_search_issues")
async def jira_search_issues(jql: str, max_results: int = 50) -> str:
try:
resources_response = await make_jira_request(
"GET",
"https://api.atlassian.com/oauth/token/accessible-resources"
)
if resources_response.status_code != 200:
return f"Failed to get accessible resources: {resources_response.status_code}"
resources = resources_response.json()
if not resources:
return "No accessible Jira resources found."
cloud_id = resources[0].get('id')
if not cloud_id:
return "Could not retrieve Cloud ID."
search_data = {
"jql": jql,
"maxResults": max_results,
"fields": ["key"]
}
search_response = await make_jira_request(
"POST",
f"https://api.atlassian.com/ex/jira/{cloud_id}/rest/api/3/search/jql",
json=search_data
)
if search_response.status_code != 200:
return f"Failed to search Jira issues: {search_response.status_code} - {search_response.text}"
search_results = search_response.json()
issues = search_results.get("issues", [])
if not issues:
return f"No issues found matching JQL: {jql}"
issue_keys = [issue["key"] for issue in issues if "key" in issue]
if not issue_keys:
return "No valid issue keys found in search results."
bulk_fetch_data = {
"issueIdsOrKeys": issue_keys,
"fields": ["key", "summary", "status", "assignee", "created", "updated"]
}
bulk_response = await make_jira_request(
"POST",
f"https://api.atlassian.com/ex/jira/{cloud_id}/rest/api/3/issue/bulkfetch",
json=bulk_fetch_data
)
if bulk_response.status_code != 200:
return f"Failed to fetch issue details: {bulk_response.status_code} - {bulk_response.text}"
bulk_results = bulk_response.json()
detailed_issues = bulk_results.get("issues", [])
issue_list = []
for issue in detailed_issues:
fields = issue.get("fields", {})
key = issue.get("key", "Unknown")
summary = fields.get("summary", "No summary")
assignee = fields.get("assignee", {}).get("displayName", "Unassigned") if fields.get("assignee") else "Unassigned"
status = fields.get("status", {}).get("name", "Unknown") if fields.get("status") else "Unknown"
issue_list.append(f"- {key}: {summary} (Status: {status}, Assignee: {assignee})")
next_page_token = search_results.get("nextPageToken")
result_text = f"Found {len(detailed_issues)} issues:\n\n" + "\n".join(issue_list)
if next_page_token:
result_text += f"\n\n[Note: More results available. Use nextPageToken: {next_page_token} for pagination]"
return result_text
except Exception as e:
logger.error(f"Error searching Jira issues: {e}")
return f"Error: {str(e)}"
@mcp.tool("jira_update_issue")
async def jira_update_issue(issue_key: str, summary: str = None, description: str = None, assignee: str = None) -> str:
try:
if not await ensure_jira_client():
return "Jira not authenticated. Please use setup_jira_oauth first."
update_data = {"fields": {}}
if summary:
update_data["fields"]["summary"] = summary
if description:
update_data["fields"]["description"] = {"type": "doc", "version": 1, "content": [{"type": "paragraph", "content": [{"type": "text", "text": description}]}]}
if assignee:
update_data["fields"]["assignee"] = {"name": assignee}
response = await jira_client.put(
f"{JIRA_URL}/rest/api/3/issue/{issue_key}",
json=update_data
)
if response.status_code == 204:
return f"Successfully updated Jira issue {issue_key}"
else:
return f"Failed to update Jira issue: {response.status_code} - {response.text}"
except Exception as e:
logger.error(f"Error updating Jira issue: {e}")
return f"Error: {str(e)}"
@mcp.tool("jira_add_comment")
async def jira_add_comment(issue_key: str, comment: str) -> str:
try:
if not await ensure_jira_client():
return "Jira not authenticated. Please use setup_jira_oauth first."
comment_data = {
"body": {
"type": "doc",
"version": 1,
"content": [{"type": "paragraph", "content": [{"type": "text", "text": comment}]}]
}
}
response = await jira_client.post(
f"{JIRA_URL}/rest/api/3/issue/{issue_key}/comment",
json=comment_data
)
if response.status_code == 201:
return f"Successfully added comment to Jira issue {issue_key}"
else:
return f"Failed to add comment: {response.status_code} - {response.text}"
except Exception as e:
logger.error(f"Error adding Jira comment: {e}")
return f"Error: {str(e)}"
@mcp.tool("jira_transition_issue")
async def jira_transition_issue(issue_key: str, transition_name: str) -> str:
try:
if not await ensure_jira_client():
return "Jira not authenticated. Please use setup_jira_oauth first."
response = await jira_client.get(
f"{JIRA_URL}/rest/api/3/issue/{issue_key}/transitions"
)
if response.status_code != 200:
return f"Failed to get transitions: {response.status_code} - {response.text}"
transitions = response.json().get("transitions", [])
target_transition = None
for transition in transitions:
if transition["name"].lower() == transition_name.lower():
target_transition = transition
break
if not target_transition:
available = [t["name"] for t in transitions]
return f"Transition '{transition_name}' not found. Available transitions: {', '.join(available)}"
transition_data = {"transition": {"id": target_transition["id"]}}
response = await jira_client.post(
f"{JIRA_URL}/rest/api/3/issue/{issue_key}/transitions",
json=transition_data
)
if response.status_code == 204:
return f"Successfully transitioned issue {issue_key} to '{transition_name}'"
else:
return f"Failed to transition issue: {response.status_code} - {response.text}"
except Exception as e:
logger.error(f"Error transitioning Jira issue: {e}")
return f"Error: {str(e)}"
@mcp.tool("setup_jira_oauth")
async def setup_jira_oauth() -> str:
try:
if not JIRA_CLIENT_ID or not JIRA_CLIENT_SECRET:
return "Jira OAuth not configured. Please set JIRA_CLIENT_ID and JIRA_CLIENT_SECRET in your configuration."
oauth_state = secrets.token_urlsafe(32)
auth_url = "https://auth.atlassian.com/authorize?" + urlencode({
"audience": "api.atlassian.com",
"client_id": JIRA_CLIENT_ID,
"scope": "read:jira-work write:jira-work read:jira-user offline_access",
"redirect_uri": "http://localhost:8080/callback",
"state": oauth_state,
"response_type": "code",
"prompt": "consent"
})
global oauth_server, oauth_code
class OAuthCallbackHandler(BaseHTTPRequestHandler):
def do_GET(self):
global oauth_code
if self.path.startswith("/callback"):
from urllib.parse import parse_qs, urlparse
query = parse_qs(urlparse(self.path).query)
print(f"[DEBUG] Callback received: {self.path}")
print(f"[DEBUG] Query parameters: {dict(query)}")
print(f"[DEBUG] Expected state: {oauth_state}")
if "code" in query and "state" in query:
code = query["code"][0]
returned_state = query["state"][0]
print(f"[DEBUG] Received state: {returned_state}")
print(f"[DEBUG] State match: {returned_state == oauth_state}")
if returned_state == oauth_state:
oauth_code = code
self.send_response(200)
self.send_header("Content-type", "text/html")
self.end_headers()
response_html = """
<html>
<head>
<title>OAuth Authorization Successful</title>
<style>
body { font-family: Arial, sans-serif; padding: 40px; text-align: center; }
.success { color: #28a745; }
.info { color: #17a2b8; }
</style>
</head>
<body>
<h2 class="success"> OAuth Authorization Successful!</h2>
<p class="info">Your Jira OAuth setup is complete.</p>
<p>You can close this window and return to your application.</p>
<p><em>This window will close automatically in 5 seconds...</em></p>
<script>setTimeout(() => window.close(), 5000);</script>
</body>
</html>
"""
self.wfile.write(response_html.encode())
threading.Thread(target=lambda: oauth_server.shutdown()).start()
else:
print(f"[ERROR] State mismatch! Expected: {oauth_state}, Got: {returned_state}")
self.send_response(400)
self.send_header("Content-type", "text/html")
self.end_headers()
error_html = f"""
<html>
<head>
<title>OAuth Error</title>
<style>
body {{ font-family: Arial, sans-serif; padding: 40px; }}
.error {{ color: #dc3545; }}
.debug {{ background: #f8f9fa; padding: 10px; margin: 10px 0; font-family: monospace; }}
</style>
</head>
<body>
<h2 class="error"> OAuth State Parameter Error</h2>
<p>The state parameter doesn't match. This is a security check to prevent CSRF attacks.</p>
<div class="debug">
<strong>Debug Information:</strong><br>
Expected state: {oauth_state}<br>
Received state: {returned_state}<br>
Match: {returned_state == oauth_state}
</div>
<p><strong>Troubleshooting:</strong></p>
<ul>
<li>Try refreshing and running the OAuth setup again</li>
<li>Make sure you're not reusing an old authorization URL</li>
<li>Clear your browser cache and cookies for this site</li>
<li>Make sure no other OAuth flows are running simultaneously</li>
</ul>
<p><a href="javascript:window.close()">Close this window</a> and try again.</p>
</body>
</html>
"""
self.wfile.write(error_html.encode())
else:
print(f"[ERROR] Missing required parameters. Query: {dict(query)}")
self.send_response(400)
self.send_header("Content-type", "text/html")
self.end_headers()
has_code = "code" in query
has_state = "state" in query
has_error = "error" in query
error_description = query.get("error_description", ["No description"])[0] if has_error else None
error_html = f"""
<html>
<head>
<title>OAuth Authorization Failed</title>
<style>
body {{ font-family: Arial, sans-serif; padding: 40px; }}
.error {{ color: #dc3545; }}
.debug {{ background: #f8f9fa; padding: 10px; margin: 10px 0; font-family: monospace; }}
.warning {{ color: #856404; background: #fff3cd; padding: 10px; border-radius: 5px; }}
</style>
</head>
<body>
<h2 class="error">❌ OAuth Authorization Failed</h2>
{f'<div class="warning"><strong>Error:</strong> {query["error"][0]}<br><strong>Description:</strong> {error_description}</div>' if has_error else ''}
<div class="debug">
<strong>Debug Information:</strong><br>
Has authorization code: {has_code}<br>
Has state parameter: {has_state}<br>
Has error parameter: {has_error}<br>
Query parameters: {dict(query)}
</div>
<p><strong>This could happen if:</strong></p>
<ul>
<li>You denied the authorization request</li>
<li>There was an error in the OAuth flow</li>
<li>The redirect URI configuration is incorrect</li>
<li>The Jira OAuth app configuration is wrong</li>
</ul>
<p><strong>Please check:</strong></p>
<ul>
<li>Your Jira OAuth app redirect URI is set to: <code>http://localhost:8080/callback</code></li>
<li>The OAuth app has the correct scopes enabled</li>
<li>The OAuth app is not in draft mode</li>
</ul>
<p><a href="javascript:window.close()">Close this window</a> and try the OAuth setup again.</p>
</body>
</html>
"""
self.wfile.write(error_html.encode())
else:
if self.path == "/":
self.send_response(200)
self.send_header("Content-type", "text/html")
self.end_headers()
debug_html = f"""
<html>
<head>
<title>Jira OAuth Debug</title>
<style>
body {{ font-family: Arial, sans-serif; padding: 40px; }}
.info {{ color: #0066cc; }}
.debug {{ background: #f8f9fa; padding: 10px; margin: 10px 0; font-family: monospace; }}
</style>
</head>
<body>
<h2 class="info">🔍 Jira OAuth Debug Information</h2>
<p>OAuth callback server is running and waiting for authorization.</p>
<div class="debug">
<strong>Configuration:</strong><br>
Expected state: {oauth_state}<br>
Redirect URI: http://localhost:8080/callback<br>
Jira URL: {JIRA_URL}<br>
Client ID: {JIRA_CLIENT_ID[:8]}...<br>
</div>
<p>Complete the authorization in the Atlassian page, then you'll be redirected to /callback</p>
</body>
</html>
"""
self.wfile.write(debug_html.encode())
else:
self.send_response(404)
self.send_header("Content-type", "text/html")
self.end_headers()
self.wfile.write(b"<html><body><h2>404 - Page not found</h2><p>Expected path: /callback</p></body></html>")
def log_message(self, format, *args):
pass
oauth_server = HTTPServer(("localhost", 8080), OAuthCallbackHandler)
try:
print(f"[INFO] Starting Jira OAuth authorization...")
print(f"[INFO] Authorization URL: {auth_url}")
print(f"[INFO] Redirect URI: http://localhost:8080/callback")
browser_opened = open_browser_with_fallback(auth_url)
if browser_opened:
print("[SUCCESS] Browser opened successfully")
print("[INFO] Please complete the authorization in your browser")
else:
print("[WARNING] Could not open browser automatically")
print("[MANUAL] Please copy and paste this URL into your browser:")
print(f"[URL] {auth_url}")
print("[INFO] After authorization, you'll be redirected to localhost:8080")
except Exception as e:
print(f"[ERROR] Error during browser opening: {e}")
print("[MANUAL] Please manually open this URL in your browser:")
print(f"[URL] {auth_url}")
print("[INFO] Starting OAuth callback server on localhost:8080...")
print("[INFO] Waiting for authorization callback...")
print("[HELP] If the browser didn't open automatically, manually visit the URL above")
print("[HELP] The authorization will redirect to http://localhost:8080/callback")
try:
oauth_server.serve_forever()
except KeyboardInterrupt:
print("[INFO] OAuth process interrupted by user")
return "[CANCELLED] OAuth setup was cancelled by user"
except Exception as e:
print(f"[ERROR] OAuth server error: {e}")
return f"[ERROR] OAuth server failed: {str(e)}"
if oauth_code:
print("[SUCCESS] Authorization code received")
print("[INFO] Exchanging authorization code for access token...")
token_data = {
"grant_type": "authorization_code",
"client_id": JIRA_CLIENT_ID,
"client_secret": JIRA_CLIENT_SECRET,
"code": oauth_code,
"redirect_uri": "http://localhost:8080/callback"
}
try:
response = httpx.post(
"https://auth.atlassian.com/oauth/token",
data=token_data,
headers={"Content-Type": "application/x-www-form-urlencoded"},
timeout=30.0
)
print(f"[DEBUG] Token exchange response status: {response.status_code}")
if response.status_code == 200:
tokens = response.json()
print("[SUCCESS] Access token obtained successfully!")
os.environ["JIRA_ACCESS_TOKEN"] = tokens["access_token"]
if "refresh_token" in tokens:
os.environ["JIRA_REFRESH_TOKEN"] = tokens["refresh_token"]
global JIRA_ACCESS_TOKEN, JIRA_REFRESH_TOKEN
JIRA_ACCESS_TOKEN = tokens["access_token"]
JIRA_REFRESH_TOKEN = tokens.get("refresh_token")
if update_env_file("JIRA_ACCESS_TOKEN", JIRA_ACCESS_TOKEN):
pass
if JIRA_REFRESH_TOKEN and update_env_file("JIRA_REFRESH_TOKEN", JIRA_REFRESH_TOKEN):
pass
print("[INFO] Jira client will be initialized with new token")
print("[INFO] Validating token and checking accessible resources...")
try:
validation_response = httpx.get(
"https://api.atlassian.com/oauth/token/accessible-resources",
headers={"Authorization": f"Bearer {JIRA_ACCESS_TOKEN}"},
timeout=10.0
)
if validation_response.status_code == 200:
resources = validation_response.json()
print(f"[SUCCESS] Token validated! Found {len(resources)} accessible resource(s)")
for resource in resources:
print(f"[INFO] - {resource.get('name', 'Unknown')}: {resource.get('url', 'No URL')}")
print(f"[INFO] Scopes: {', '.join(resource.get('scopes', []))}")
else:
print(f"[WARNING] Token validation failed: {validation_response.status_code}")
except Exception as e:
print(f"[WARNING] Could not validate token: {e}")
success_msg = f"""[SUCCESS] Jira OAuth setup completed successfully!
✅ Access token obtained and configured
✅ Tokens automatically saved to .env file
✅ Jira tools are now available for use
Tokens have been saved to your .env file and will persist across sessions.
You can now use all Jira tools like jira_create_issue, jira_search_issues, etc.
To manually check your tokens in .env:
JIRA_ACCESS_TOKEN={JIRA_ACCESS_TOKEN[:20]}..."""
if JIRA_REFRESH_TOKEN:
success_msg += f"\n JIRA_REFRESH_TOKEN={JIRA_REFRESH_TOKEN[:20]}..."
return success_msg
else:
error_response = response.text
print(f"[ERROR] Token exchange failed: {response.status_code}")
print(f"[ERROR] Response: {error_response}")
return f"[ERROR] Failed to exchange authorization code for tokens\nStatus: {response.status_code}\nResponse: {error_response}\n\nThis could be due to:\n- Invalid client credentials\n- Expired authorization code\n- Incorrect redirect URI\n\nPlease check your JIRA_CLIENT_ID and JIRA_CLIENT_SECRET configuration."
except httpx.TimeoutException:
return "[ERROR] Token exchange timed out. Please check your internet connection and try again."
except Exception as e:
print(f"[ERROR] Exception during token exchange: {e}")
return f"[ERROR] Token exchange failed with exception: {str(e)}"
else:
print("[ERROR] No authorization code received")
return "[ERROR] OAuth authorization failed or was cancelled\n\nPossible causes:\n- User denied authorization\n- OAuth flow was interrupted\n- Browser/network issues\n\nPlease try running setup_jira_oauth again."
except Exception as e:
logger.error(f"Error setting up Jira OAuth: {e}")
return f"Error: {str(e)}"
@mcp.tool("check_github_permissions")
async def check_github_permissions() -> str:
try:
current_user = await get_github_user()
if not current_user:
return "No GitHub token configured or invalid token"
response = await github_client.get(f"https://api.github.com/users/{current_user}/repos")
if response.status_code == 200:
repos = response.json()
return f"GitHub user: {current_user}\nAccess level: Authorized\nRepositories: {len(repos)} accessible\n\nYou can:\n- Read files from public repositories\n- Read files from your own repositories\n- Create issues in your own repositories\n- Create pull requests in your own repositories\n\nYou cannot:\n- Create new repositories\n- Delete repositories\n- Access private repositories of other users\n- Create issues/PRs in other users' repositories"
else:
return f"GitHub user: {current_user}\nAccess level: Limited\nStatus: {response.status_code}"
except Exception as e:
logger.error(f"Error checking GitHub permissions: {e}")
return f"Error: {str(e)}"
@mcp.tool("check_jira_permissions")
async def check_jira_permissions() -> str:
try:
try:
resources_response = await make_jira_request(
"GET",
"https://api.atlassian.com/oauth/token/accessible-resources"
)
if resources_response.status_code != 200:
return f"Failed to get accessible resources: {resources_response.status_code}"
resources = resources_response.json()
if not resources:
return "No accessible Jira resources found. Please check your OAuth app configuration."
result = f"Jira OAuth Status: Authenticated\n\nAccessible Resources ({len(resources)}):"
for resource in resources:
cloudid = resource.get('id', 'Unknown')
name = resource.get('name', 'Unknown')
url = resource.get('url', 'No URL')
scopes = resource.get('scopes', [])
result += f"\n\n📍 Site: {name}\n URL: {url}\n Cloud ID: {cloudid}\n Scopes: {', '.join(scopes)}"
try:
user_response = await make_jira_request(
"GET",
f"https://api.atlassian.com/ex/jira/{cloudid}/rest/api/3/myself"
)
if user_response.status_code == 200:
user_info = user_response.json()
result += f"\n API Access: Working\n User: {user_info.get('displayName', 'Unknown')} ({user_info.get('emailAddress', 'No email')})"
else:
result += f"\n API Access: Failed ({user_response.status_code})"
except Exception as e:
result += f"\n API Access: Error ({str(e)})"
result += "\n\n🔧 Available Actions:\n"
result += "- Create issues in any accessible project\n"
result += "- Search and update existing issues\n"
result += "- Add comments and transition issues\n"
result += "- Access project and user information\n"
return result
except Exception as e:
return f"Error checking accessible resources: {str(e)}"
except Exception as e:
logger.error(f"Error checking Jira permissions: {e}")
return f"Error: {str(e)}"
@mcp.tool("refresh_jira_token")
async def refresh_jira_token() -> str:
"""Refresh the Jira access token using the refresh token."""
global JIRA_ACCESS_TOKEN, JIRA_REFRESH_TOKEN, jira_client
try:
if not JIRA_REFRESH_TOKEN:
return "No refresh token available. Please run setup_jira_oauth again."
if not JIRA_CLIENT_ID or not JIRA_CLIENT_SECRET:
return "Jira OAuth not configured. Please set JIRA_CLIENT_ID and JIRA_CLIENT_SECRET."
print("[INFO] Refreshing Jira access token...")
refresh_data = {
"grant_type": "refresh_token",
"client_id": JIRA_CLIENT_ID,
"client_secret": JIRA_CLIENT_SECRET,
"refresh_token": JIRA_REFRESH_TOKEN
}
try:
response = httpx.post(
"https://auth.atlassian.com/oauth/token",
data=refresh_data,
headers={"Content-Type": "application/x-www-form-urlencoded"},
timeout=30.0
)
if response.status_code == 200:
tokens = response.json()
JIRA_ACCESS_TOKEN = tokens["access_token"]
if "refresh_token" in tokens:
JIRA_REFRESH_TOKEN = tokens["refresh_token"]
os.environ["JIRA_REFRESH_TOKEN"] = JIRA_REFRESH_TOKEN
os.environ["JIRA_ACCESS_TOKEN"] = JIRA_ACCESS_TOKEN
update_env_file("JIRA_ACCESS_TOKEN", JIRA_ACCESS_TOKEN)
if "refresh_token" in tokens:
update_env_file("JIRA_REFRESH_TOKEN", JIRA_REFRESH_TOKEN)
jira_client = None
print("[SUCCESS] Access token refreshed successfully!")
return f"""[SUCCESS] Jira access token refreshed!
New access token obtained
Token expires in: {tokens.get('expires_in', 'Unknown')} seconds
Jira tools are ready to use
To make this permanent, update your .env file:
JIRA_ACCESS_TOKEN={JIRA_ACCESS_TOKEN[:20]}..."""
else:
error_response = response.text
if "invalid_grant" in error_response:
return """[ERROR] Refresh token is invalid or expired.
Possible causes:
- Refresh token has expired (90 days of inactivity)
- User password was changed
- App authorization was revoked
Solution: Run setup_jira_oauth again to get new tokens."""
else:
return f"[ERROR] Token refresh failed: {response.status_code}\nResponse: {error_response}"
except Exception as e:
return f"[ERROR] Token refresh failed with exception: {str(e)}"
except Exception as e:
return f"Error refreshing token: {str(e)}"
@mcp.tool("get_jira_cloudid")
async def get_jira_cloudid() -> str:
"""Get the cloudid for your Jira instance, needed for direct API calls."""
try:
response = await make_jira_request(
"GET",
"https://api.atlassian.com/oauth/token/accessible-resources"
)
if response.status_code == 200:
resources = response.json()
if not resources:
return "No accessible Jira resources found."
result = "Jira Cloud IDs for API Access:\n\n"
for resource in resources:
cloudid = resource.get('id', 'Unknown')
name = resource.get('name', 'Unknown')
url = resource.get('url', 'No URL')
result += f"Site: {name}\n"
result += f"URL: {url}\n"
result += f"Cloud ID: {cloudid}\n"
result += f"API Base: https://api.atlassian.com/ex/jira/{cloudid}/\n\n"
result += "💡 Usage Tips:\n"
result += "- Use the Cloud ID for direct API calls to api.atlassian.com\n"
result += "- Replace {cloudid} in API documentation examples\n"
result += "- All requests must include 'Authorization: Bearer <access_token>' header\n"
return result
else:
return f"Failed to get accessible resources: {response.status_code}"
except Exception as e:
return f"Error getting cloudid: {str(e)}"
@mcp.tool("test_connection")
async def test_connection() -> str:
try:
result = "[SUCCESS] MCP Server is working!\n\n"
if GITHUB_TOKEN:
result += "GitHub: Token configured\n"
else:
result += "GitHub: No token configured\n"
if JIRA_URL and JIRA_CLIENT_ID and JIRA_CLIENT_SECRET:
result += "Jira: OAuth credentials configured\n"
if JIRA_ACCESS_TOKEN:
result += "Jira: Access token available\n"
else:
result += "Jira: Access token needed - run setup_jira_oauth\n"
else:
result += "Jira: OAuth credentials not configured\n"
result += "\nAvailable tools:\n"
result += "- GitHub tools: Read files, create issues/PRs\n"
result += "- Jira tools: Create/search/update issues (after OAuth)\n"
return result
except Exception as e:
return f"Error testing connection: {e}"
@mcp.tool("check_browser_availability")
async def check_browser_availability() -> str:
"""Check what browsers are available on the system."""
try:
result = "[INFO] Browser Availability Check:\n\n"
try:
browsers = webbrowser._browsers
if browsers:
result += "[SUCCESS] Webbrowser module found browsers:\n"
for name, browser in browsers.items():
result += f" - {name}: {browser}\n"
else:
result += "[ERROR] No browsers found in webbrowser module\n"
except Exception as e:
result += f"[ERROR] Error checking webbrowser module: {e}\n"
if platform.system() == "Windows":
result += "\n[INFO] Windows-specific checks:\n"
try:
subprocess.run(["start", "/?"], shell=True, capture_output=True, check=True)
result += "[SUCCESS] 'start' command available\n"
except Exception:
result += "[ERROR] 'start' command not available\n"
try:
subprocess.run(["explorer", "/?"], shell=True, capture_output=True, check=True)
result += "[SUCCESS] 'explorer' command available\n"
except Exception:
result += "[ERROR] 'explorer' command not available\n"
result += "\n[INFO] Testing URL opening:\n"
test_url = "https://www.google.com"
try:
opened = open_browser_with_fallback(test_url)
if opened:
result += f"[SUCCESS] Successfully opened test URL: {test_url}\n"
else:
result += f"[ERROR] Failed to open test URL: {test_url}\n"
except Exception as e:
result += f"[ERROR] Error testing URL opening: {e}\n"
return result
except Exception as e:
return f"Error checking browser availability: {e}"
@mcp.prompt("check_repository_security")
async def check_repository_security() -> Prompt:
return Prompt(
description="Check repository security permissions and access levels",
arguments=[
PromptArgument(name="repository", description="Repository to check (format: owner/repo)"),
]
)
@mcp.prompt("read_github_file")
async def read_github_file() -> Prompt:
return Prompt(
description="Read a file from a GitHub repository",
arguments=[
PromptArgument(name="repository", description="Repository to read from (format: owner/repo)"),
PromptArgument(name="file_path", description="Path to the file in the repository"),
]
)
@mcp.prompt("create_github_issue")
async def create_github_issue() -> Prompt:
return Prompt(
description="Create a new issue in a GitHub repository",
arguments=[
PromptArgument(name="repository", description="Repository to create issue in (format: owner/repo)"),
PromptArgument(name="title", description="Issue title"),
PromptArgument(name="body", description="Issue description"),
PromptArgument(name="labels", description="Optional labels (comma-separated)"),
]
)
@mcp.prompt("create_jira_issue")
async def create_jira_issue() -> Prompt:
return Prompt(
description="Create a new issue in a Jira project",
arguments=[
PromptArgument(name="project_key", description="Jira project key"),
PromptArgument(name="summary", description="Issue summary"),
PromptArgument(name="description", description="Issue description"),
PromptArgument(name="issue_type", description="Issue type (e.g., Task, Bug, Story)"),
]
)
@mcp.prompt("search_jira_issues")
async def search_jira_issues() -> Prompt:
return Prompt(
description="Search for issues in Jira using JQL",
arguments=[
PromptArgument(name="jql", description="JQL search query"),
PromptArgument(name="max_results", description="Maximum number of results (default: 50)"),
]
)
@mcp.prompt("setup_jira_oauth_authentication")
async def setup_jira_oauth_authentication() -> Prompt:
return Prompt(
description="Set up Jira OAuth authentication to enable Jira tools",
arguments=[]
)
if __name__ == "__main__":
if not GITHUB_TOKEN:
logger.warning("GitHub token not configured. GitHub features will be limited.")
if not JIRA_URL or not JIRA_CLIENT_ID or not JIRA_CLIENT_SECRET:
logger.warning("Jira OAuth not configured. Jira features will be unavailable.")
logger.info("Starting GitHub-Jira MCP Server...")
logger.info("Server will wait for MCP protocol messages...")
try:
mcp.run(transport='stdio')
except Exception as e:
logger.error(f"Server error: {e}")
import traceback
traceback.print_exc()
sys.exit(1)