JiraAgent

class JiraAgent(Agent):
    """Intelligent Jira agent with automatic configuration discovery."""

    def __init__(
        self,
        jira_config: Dict[str, Any] = None,
        max_steps: int = 10,
        model_id: str = "Qwen3-Coder-30B-A3B-Instruct-GGUF",
        silent_mode: bool = False,
        **kwargs
    ):
        """
        Initialize Jira agent.

        Args:
            jira_config: Pre-discovered Jira configuration {
                "projects": [{"key": "PROJ", "name": "Project Name"}],
                "issue_types": ["Bug", "Task", "Story"],
                "statuses": ["To Do", "In Progress", "Done"],
                "priorities": ["Highest", "High", "Medium", "Low"]
            }
        """
        pass

    def initialize(self) -> Dict[str, Any]:
        """
        Discover and cache Jira configuration.

        Returns:
            Discovered configuration dict

        Raises:
            Exception: If credentials invalid or API fails
        """
        pass

    def get_config(self) -> Optional[Dict[str, Any]]:
        """Get cached Jira configuration."""
        pass

    # Tools
    @tool
    def jira_search(
        jql: str = "created >= -30d ORDER BY updated DESC",
        max_results: int = None,
        fields: str = None
    ) -> Dict[str, Any]:
        """Search Jira issues using JQL query."""
        pass

    @tool
    def jira_create(
        summary: str,
        description: str = "",
        issue_type: str = "Task",
        priority: str = None,
        project: str = None
    ) -> Dict[str, Any]:
        """Create new Jira issue."""
        pass

    @tool
    def jira_update(
        issue_key: str,
        summary: str = None,
        description: str = None,
        priority: str = None,
        status: str = None
    ) -> Dict[str, Any]:
        """Update existing Jira issue."""
        pass

# Required
ATLASSIAN_SITE_URL=https://company.atlassian.net
ATLASSIAN_API_KEY=your_api_token_here
ATLASSIAN_USER_EMAIL=[email protected]

async def _discover_jira_config(self) -> Dict[str, Any]:
    """Discover Jira instance configuration via API."""

    site_url, api_key, user_email = self._get_jira_credentials()
    auth_header = base64.b64encode(f"{user_email}:{api_key}".encode()).decode()

    headers = {
        "Authorization": f"Basic {auth_header}",
        "Content-Type": "application/json"
    }

    config = {
        "projects": [],
        "issue_types": [],
        "statuses": [],
        "priorities": []
    }

    async with aiohttp.ClientSession() as session:
        # Get projects
        async with session.get(f"{site_url}/rest/api/3/project", headers=headers) as resp:
            projects = await resp.json()
            config["projects"] = [{"key": p["key"], "name": p["name"]} for p in projects]

        # Get issue types (exclude subtasks)
        async with session.get(f"{site_url}/rest/api/3/issuetype", headers=headers) as resp:
            types = await resp.json()
            config["issue_types"] = [t["name"] for t in types if not t.get("subtask")]

        # Get statuses
        async with session.get(f"{site_url}/rest/api/3/status", headers=headers) as resp:
            statuses = await resp.json()
            config["statuses"] = [s["name"] for s in statuses]

        # Get priorities
        async with session.get(f"{site_url}/rest/api/3/priority", headers=headers) as resp:
            priorities = await resp.json()
            config["priorities"] = [p["name"] for p in priorities]

    return config

def _get_system_prompt(self) -> str:
    """Generate system prompt with discovered configuration."""

    prompt = """You are a Jira assistant that responds ONLY in JSON format.

**JQL Basics:**
- Current user: assignee = currentUser()
- By key: key = "PROJ-123"
- Operators: AND, OR, NOT"""

    # Add discovered configuration
    if self._jira_config:
        if self._jira_config.get("projects"):
            keys = [p["key"] for p in self._jira_config["projects"]]
            prompt += f"\n- Projects: {', '.join(keys)}"

        if self._jira_config.get("issue_types"):
            prompt += f"\n- Types: {', '.join(self._jira_config['issue_types'])}"

        if self._jira_config.get("priorities"):
            prompt += f"\n- Priorities: {', '.join(self._jira_config['priorities'])}"

        if self._jira_config.get("statuses"):
            prompt += f"\n- Statuses: {', '.join(self._jira_config['statuses'])}"

    prompt += """

**JQL QUOTING RULES:**
- Single words: NO quotes → status = Done
- Multiple words: DOUBLE quotes → status = "In Progress"
- NEVER use single quotes
- Functions: NO quotes → assignee = currentUser()"""

    return prompt

async def _execute_jira_search_async(
    self,
    jql: str,
    max_results: int = None,
    fields: str = None
) -> Dict[str, Any]:
    """Execute Jira search API call."""

    site_url, api_key, user_email = self._get_jira_credentials()
    auth_header = base64.b64encode(f"{user_email}:{api_key}".encode()).decode()

    headers = {
        "Authorization": f"Basic {auth_header}",
        "Content-Type": "application/json"
    }

    params = {"jql": jql}
    if max_results:
        params["maxResults"] = max_results
    if fields:
        params["fields"] = fields
    else:
        params["fields"] = "key,summary,status,priority,issuetype,assignee"

    async with aiohttp.ClientSession() as session:
        url = f"{site_url}/rest/api/3/search/jql"
        async with session.get(url, headers=headers, params=params) as response:
            response.raise_for_status()
            data = await response.json()

            issues = []
            for issue in data.get("issues", []):
                fields = issue.get("fields", {})
                issues.append({
                    "key": issue.get("key"),
                    "summary": fields.get("summary"),
                    "status": fields.get("status", {}).get("name"),
                    "priority": fields.get("priority", {}).get("name"),
                    "type": fields.get("issuetype", {}).get("name"),
                    "assignee": fields.get("assignee", {}).get("displayName", "Unassigned")
                })

            return {
                "issues": issues,
                "total": data.get("total", len(issues)),
                "jql": jql
            }

# tests/agents/test_jira_agent.py

def test_jira_agent_initialization():
    """Test JiraAgent initializes."""
    config = {
        "projects": [{"key": "TEST", "name": "Test Project"}],
        "issue_types": ["Bug", "Task"],
        "statuses": ["To Do", "Done"],
        "priorities": ["High", "Low"]
    }
    agent = JiraAgent(jira_config=config, silent_mode=True)
    assert agent.get_config() == config

@pytest.mark.asyncio
async def test_jira_search():
    """Test JQL search execution."""
    agent = JiraAgent(silent_mode=True)
    # Mock aiohttp session
    result = await agent._execute_jira_search_async("assignee = currentUser()")
    assert "issues" in result
    assert "total" in result
    assert "jql" in result

def test_dynamic_system_prompt():
    """Test system prompt includes discovered config."""
    config = {
        "projects": [{"key": "PROJ", "name": "Project"}],
        "issue_types": ["Bug"],
        "priorities": ["High"]
    }
    agent = JiraAgent(jira_config=config, silent_mode=True)
    prompt = agent._get_system_prompt()

    assert "PROJ" in prompt
    assert "Bug" in prompt
    assert "High" in prompt

# Set credentials
export ATLASSIAN_SITE_URL=https://company.atlassian.net
export ATLASSIAN_API_KEY=your_token
export ATLASSIAN_USER_EMAIL=you@company.com

# Use agent
gaia jira "Show me high priority bugs assigned to me"

from gaia.agents.jira.agent import JiraAgent

# Initialize and discover
agent = JiraAgent()
config = agent.initialize()

print(f"Found {len(config['projects'])} projects")

# Natural language query
result = agent.process_query("Show my open issues")
print(result["result"])

# Use pre-discovered config (faster initialization)
config = {
    "projects": [{"key": "GAIA", "name": "GAIA Project"}],
    "issue_types": ["Bug", "Task", "Story"],
    "statuses": ["To Do", "In Progress", "Done"],
    "priorities": ["Highest", "High", "Medium", "Low"]
}

agent = JiraAgent(jira_config=config)
result = agent.process_query("Create a bug: Login fails")