The main file. It defines the MCP connection, all four agents, all four tasks, and the crew orchestration.
"""
CrewAI + Jira MCP: Auto-Generate Test Plans, Test Cases & Playwright Scripts
Input : A Jira ticket ID (e.g., VWO-48)
Output : test_plan.md, test_cases.md, playwright_tests.md
Pipeline:
1. Jira Analyst -> fetches ticket via MCP, extracts requirements
2. Test Plan Writer -> writes complete test plan (12 sections)
3. Test Case Writer -> writes detailed test cases table
4. Playwright Coder -> generates automation scripts
"""
import os
import datetime
from crewai import Agent, Task, Crew, Process, LLM
from crewai_tools import MCPServerAdapter
from mcp import StdioServerParameters
from dotenv import load_dotenv
load_dotenv()
# ================= LLM SETUP =================
llm = LLM(
model="groq/llama-3.3-70b-versatile",
api_key=os.getenv("GROQ_API_KEY"),
num_retries=5,
request_timeout=120,
)
# If using OpenAI instead, uncomment:
# llm = LLM(model="openai/gpt-4o-mini", api_key=os.getenv("OPENAI_API_KEY"))
# ============ MCP SERVER CONFIGURATION ============
def get_mcp_server_params() -> StdioServerParameters:
"""
Configure the mcp-atlassian server connection.
Launches `uvx mcp-atlassian` as a subprocess and talks over STDIO.
The MCP server handles: auth, ADF -> text, pagination, retries.
"""
return StdioServerParameters(
command="uvx",
args=["mcp-atlassian"],
env={
"JIRA_URL": os.getenv("JIRA_URL"),
"JIRA_USERNAME": os.getenv("JIRA_USERNAME"),
"JIRA_API_TOKEN": os.getenv("JIRA_API_TOKEN"),
"PATH": os.environ.get("PATH", ""),
"UV_PYTHON": "3.12",
},
)
# ============ TEST PLAN TEMPLATE ============
TEST_PLAN_TEMPLATE = """
Follow this EXACT structure for the test plan:
### 1. TEST PLAN OVERVIEW (ID: TP-<JIRA_ID>, project, author, date, version)
### 2. OBJECTIVE ### 3. SCOPE (in / out)
### 4. TEST STRATEGY (types, approach, tool = Playwright + TypeScript)
### 5. TEST ENVIRONMENT ### 6. ENTRY CRITERIA ### 7. EXIT CRITERIA
### 8. TEST SCENARIOS (high level) ### 9. RISK ASSESSMENT
### 10. DEFECT MANAGEMENT ### 11. TEST SCHEDULE ### 12. SIGN-OFF
"""
# ================= AGENTS =================
def create_agents(mcp_tools: list, ticket_id: str):
"""Create all four agents with MCP tools and return them."""
jira_analyst = Agent(
role="Senior QA Analyst",
goal=(
f"Fetch Jira ticket {ticket_id} using the available Jira tools, "
"then extract ALL testable requirements, acceptance criteria, "
"edge cases, and risks."
),
backstory=(
"You are a senior QA analyst with 12+ years of experience. "
"You have access to Jira through MCP tools. Fetch the ticket "
"first, then analyze functional requirements, acceptance "
"criteria, edge cases, boundary conditions, and risks."
),
tools=mcp_tools, # <- MCP tools injected here!
llm=llm,
verbose=True,
allow_delegation=False,
max_iter=10,
)
test_plan_writer = Agent(
role="Test Plan Documentation Specialist",
goal="Create a comprehensive 12-section test plan document.",
backstory=(
f"You are a certified ISTQB test planning expert. "
f"You MUST follow this template:\n{TEST_PLAN_TEMPLATE}\n"
f"Today is {datetime.date.today().strftime('%B %d, %Y')}. "
f"Use professional markdown formatting."
),
llm=llm, verbose=True, allow_delegation=False, max_iter=10,
)
test_case_writer = Agent(
role="Test Case Design Specialist",
goal="Design detailed, executable test cases (positive, negative, edge, boundary).",
backstory=(
"You write test cases so clear that any junior tester can "
"execute them: TC ID, Title, Preconditions, Steps, Expected "
"Results, Test Data, Priority. You cover happy path, negative, "
"edge, boundary, UI validation, and error handling."
),
llm=llm, verbose=True, allow_delegation=False, max_iter=10,
)
playwright_coder = Agent(
role="Playwright Automation Engineer",
goal="Generate production-ready Playwright tests in TypeScript.",
backstory=(
"You are a senior SDET. Your scripts follow best practices: "
"Page Object Model, data-testid > CSS > XPath locators, "
"test.describe/test(), expect() assertions, beforeEach/"
"afterEach, and clear comments. You output COMPLETE, RUNNABLE files."
),
llm=llm, verbose=True, allow_delegation=False, max_iter=10,
)
return jira_analyst, test_plan_writer, test_case_writer, playwright_coder
# ================= TASKS =================
def create_tasks(agents: tuple, ticket_id: str):
"""Create all four tasks and wire them together."""
jira_analyst, test_plan_writer, test_case_writer, playwright_coder = agents
analysis_task = Task(
description=(
f"Fetch Jira ticket {ticket_id} using the Jira MCP tools "
"(look for jira_get_issue or similar). Then provide a DETAILED "
"analysis: summary, testable requirements, acceptance criteria, "
"edge cases, risks, and recommended testing types."
),
expected_output="A detailed analysis report: summary, requirements, AC, edge cases, risks, testing types.",
agent=jira_analyst,
)
test_plan_task = Task(
description=(
f"Based on the analysis, create a COMPLETE test plan for {ticket_id}. "
"Follow the 12-section template EXACTLY. Include high-level scenarios, "
"3-5 risks, and a test schedule. Clean markdown."
),
expected_output="A complete 12-section test plan in markdown.",
agent=test_plan_writer,
context=[analysis_task],
output_file="output/test_plan.md",
)
test_cases_task = Task(
description=(
f"Based on the analysis and test plan, create 12-15 DETAILED test "
"cases for {ticket_id} in a markdown table: TC ID | Title | "
"Preconditions | Steps | Expected Result | Test Data | Priority. "
"Cover positive, negative, edge, UI, and API scenarios."
),
expected_output="A markdown document with 12-15 detailed test cases in table format.",
agent=test_case_writer,
context=[analysis_task, test_plan_task],
output_file="output/test_cases.md",
)
playwright_task = Task(
description=(
f"Based on the test cases, generate COMPLETE Playwright TypeScript "
"scripts for {ticket_id}: Page Object Model, data-testid locators, "
"test.describe() groups, expect() assertions, beforeEach setup, "
"helpful comments. No placeholders or TODOs."
),
expected_output="Complete, copy-paste-ready Playwright TypeScript test files.",
agent=playwright_coder,
context=[analysis_task, test_cases_task],
output_file="output/playwright_tests.md",
)
return analysis_task, test_plan_task, test_cases_task, playwright_task
# ============ CREW ORCHESTRATION ============
def run_crew(ticket_id: str):
"""Connect to Jira MCP -> create Crew -> run pipeline."""
print(f"\n[*] Target Jira Ticket: {ticket_id}")
os.makedirs("output", exist_ok=True)
server_params = get_mcp_server_params()
print("[*] Connecting to Jira MCP server (mcp-atlassian)...")
with MCPServerAdapter(server_params, connect_timeout=60) as mcp_tools:
tool_names = [t.name for t in mcp_tools]
print(f"[+] Connected! Discovered {len(mcp_tools)} Jira tools:")
for name in tool_names:
print(f" - {name}")
agents = create_agents(mcp_tools, ticket_id)
tasks = create_tasks(agents, ticket_id)
crew = Crew(
agents=list(agents),
tasks=list(tasks),
process=Process.sequential,
verbose=True,
max_rpm=4, # Rate limit for Groq free tier
)
print(f"\n[*] Starting QA Pipeline for {ticket_id}")
result = crew.kickoff()
print("\n[+] QA PIPELINE COMPLETE! Files in ./output/")
return result