model context protocol

# jira_mcp_server.py
import asyncio
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
 
server = Server("jira-mcp")
 
@server.list_tools()
async def list_tools():
   return [
       Tool(
           name="get_issue",
           description="Get detailed information about a Jira issue by its key (e.g., PROJ-123)",
           inputSchema={
               "type": "object",
               "properties": {
                   "issue_key": {
                       "type": "string",
                       "description": "The Jira issue key (e.g., 'PROJ-123')"
                   }
               },
               "required": ["issue_key"]
           }
       ),
       Tool(
           name="create_issue",
           description="Create a new Jira issue in a project",
           inputSchema={
               "type": "object",
               "properties": {
                   "project_key": {"type": "string", "description": "Project key (e.g., 'PROJ')"},
                   "summary": {"type": "string", "description": "Issue title"},
                   "issue_type": {"type": "string", "description": "Type: Bug, Task, Story"},
                   "description": {"type": "string", "description": "Detailed description"}
               },
               "required": ["project_key", "summary", "issue_type"]
           }
       ),
       Tool(
           name="search_issues",
           description="Search Jira issues using JQL query",
           inputSchema={
               "type": "object",
               "properties": {
                   "jql": {"type": "string", "description": "JQL query string"},
                   "max_results": {"type": "integer", "default": 20}
               },
               "required": ["jql"]
           }
       )
   ]
 
@server.call_tool()
async def call_tool(name: str, arguments: dict):
   if name == "get_issue":
       issue = await jira_client.get_issue(arguments["issue_key"])
       return [TextContent(
           type="text",
           text=f"**{issue['key']}**: {issue['summary']}\n"
                f"Status: {issue['status']}\n"
                f"Assignee: {issue['assignee']}\n"
                f"Description: {issue['description']}"
       )]
 
   elif name == "create_issue":
       result = await jira_client.create_issue(
           project=arguments["project_key"],
           summary=arguments["summary"],
           issue_type=arguments["issue_type"],
           description=arguments.get("description", "")
       )
       return [TextContent(type="text", text=f"Created issue: {result['key']}")]
 
   elif name == "search_issues":
       issues = await jira_client.search(arguments["jql"], arguments.get("max_results", 20))
       formatted = "\n".join([f"- {i['key']}: {i['summary']} [{i['status']}]" for i in issues])
       return [TextContent(type="text", text=formatted)]
 
   raise ValueError(f"Unknown tool: {name}")
 
 
# Chạy server qua stdio
async def main():
   async with stdio_server() as (read_stream, write_stream):
       await server.run(read_stream, write_stream, server.create_initialization_options())
 
if __name__ == "__main__":
   asyncio.run(main())

# jira_mcp_server.py (mcp 2.x)
from mcp.server.mcpserver import MCPServer
 
mcp = MCPServer("jira-mcp")
 
@mcp.tool()
async def get_issue(issue_key: str) -> str:
   """Get detailed information about a Jira issue by its key (e.g., PROJ-123)"""
   issue = await jira_client.get_issue(issue_key)
   return f"**{issue['key']}**: {issue['summary']}\nStatus: {issue['status']}"
 
@mcp.tool()
async def search_issues(jql: str, max_results: int = 20) -> str:
   """Search Jira issues using JQL query"""
   issues = await jira_client.search(jql, max_results)
   return "\n".join(f"- {i['key']}: {i['summary']} [{i['status']}]" for i in issues)
 
if __name__ == "__main__":
   mcp.run()  # mặc định transport stdio

// .kiro/settings/mcp.json hoặc tương đương
{
  "mcpServers": {
    "jira": {
      "command": "uvx",
      "args": ["jira-mcp-server@latest"],
      "env": {
        "JIRA_URL": "https://your-company.atlassian.net",
        "JIRA_TOKEN": "${JIRA_API_TOKEN}"
      }
    },
    "confluence": {
      "command": "uvx",
      "args": ["confluence-mcp-server@latest"],
      "env": {
        "CONFLUENCE_URL": "https://your-company.atlassian.net/wiki",
        "CONFLUENCE_TOKEN": "${CONFLUENCE_TOKEN}"
      }
    },
    "postgres": {
      "command": "uvx",
      "args": ["postgres-mcp-server@latest"],
      "env": {
        "DATABASE_URL": "postgresql://user:pass@localhost:5432/mydb"
      }
    }
  }
}

@server.list_resources()
async def list_resources():
   return [
       Resource(
           uri="jira://project/PORTAL/issues",
           name="PORTAL Project Issues",
           description="All open issues in PORTAL project",
           mimeType="application/json"
       ),
       Resource(
           uri="confluence://space/ENG/pages",
           name="Engineering Space Pages",
           description="All pages in Engineering Confluence space"
       )
   ]
 
@server.read_resource()
async def read_resource(uri):
   # SDK truyền vào kiểu AnyUrl chứ không phải str, so sánh trực tiếp với chuỗi sẽ luôn False
   if str(uri) == "jira://project/PORTAL/issues":
       issues = await jira_client.get_open_issues("PORTAL")
       return formatted_issues(issues)
   raise ValueError(f"Unknown resource: {uri}")

@server.list_prompts()
async def list_prompts():
   return [
       Prompt(
           name="review-pr",
           description="Review a pull request for code quality and potential issues",
           arguments=[
               PromptArgument(name="pr_number", description="PR number to review", required=True)
           ]
       )
   ]
 
@server.get_prompt()
async def get_prompt(name: str, arguments: dict):
   if name == "review-pr":
       pr_data = await github_client.get_pr(arguments["pr_number"])
       return GetPromptResult(
           messages=[
               PromptMessage(
                   role="user",
                   content=TextContent(
                       type="text",
                       text=f"Review this PR for code quality:\n\n"
                            f"Title: {pr_data['title']}\n"
                            f"Files changed: {pr_data['files']}\n"
                            f"Diff:\n{pr_data['diff']}\n\n"
                            f"Focus on: security, performance, readability."
                   )
               )
           ]
       )

import re
 
@server.call_tool()
async def call_tool(name: str, arguments: dict):
   # Always validate
   if name == "delete_issue":
       # str không có method .match(), phải dùng module re
       if not re.fullmatch(r"[A-Z][A-Z0-9]*-\d+", arguments.get("issue_key", "")):
           raise ValueError("Invalid issue key format")
 
       # Log for audit (user lấy từ thông tin xác thực của request, tuỳ cách bạn làm auth)
       logger.info(f"TOOL_CALL: delete_issue | key={arguments['issue_key']} | user={current_user}")
 
       # Execute
       await jira_client.delete_issue(arguments["issue_key"])
       return [TextContent(type="text", text=f"Deleted {arguments['issue_key']}")]