88 lines
2.8 KiB
Python
88 lines
2.8 KiB
Python
|
"""
|
||
|
|
title: Aha! Pipeline
|
||
|
|
author: PM Factory
|
||
|
|
version: 0.1.0
|
||
|
|
description: Create epics and features in Aha! via the MCP server.
|
||
|
|
requirements: subprocess
|
||
|
|
"""
|
||
|
|
|
||
|
|
import subprocess
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
from typing import Optional
|
||
|
|
|
||
|
|
MCPORTER_CONFIG = os.environ.get("MCPORTER_CONFIG", "config/mcporter.json")
|
||
|
|
|
||
|
|
|
||
|
|
class Tools:
|
||
|
|
def __init__(self):
|
||
|
|
self.valves = self.Valves()
|
||
|
|
|
||
|
|
class Valves:
|
||
|
|
MCPORTER_CONFIG: str = MCPORTER_CONFIG
|
||
|
|
AHA_DOMAIN: str = os.environ.get("AHA_DOMAIN", "")
|
||
|
|
AHA_API_KEY: str = os.environ.get("AHA_API_KEY", "")
|
||
|
|
|
||
|
|
def aha_create_epic(self, product: str, name: str, description: str, workflow_status: str = "New", __user__: dict = {}) -> str:
|
||
|
|
"""
|
||
|
|
Create an epic in Aha!
|
||
|
|
|
||
|
|
:param product: Aha! product key (e.g. 'PLAT')
|
||
|
|
:param name: Epic name
|
||
|
|
:param description: Epic description (supports HTML)
|
||
|
|
:param workflow_status: Initial status. Default: New
|
||
|
|
:return: Created epic reference and URL
|
||
|
|
"""
|
||
|
|
env = os.environ.copy()
|
||
|
|
env["AHA_DOMAIN"] = self.valves.AHA_DOMAIN
|
||
|
|
env["AHA_API_TOKEN"] = self.valves.AHA_API_KEY
|
||
|
|
|
||
|
|
params = {
|
||
|
|
"product": product,
|
||
|
|
"name": name,
|
||
|
|
"description": description,
|
||
|
|
"workflow_status": workflow_status
|
||
|
|
}
|
||
|
|
try:
|
||
|
|
result = subprocess.run(
|
||
|
|
[
|
||
|
|
"mcporter",
|
||
|
|
"--config", self.valves.MCPORTER_CONFIG,
|
||
|
|
"call", "aha", "create_epic",
|
||
|
|
"--params", json.dumps(params)
|
||
|
|
],
|
||
|
|
capture_output=True, text=True, timeout=30, env=env
|
||
|
|
)
|
||
|
|
if result.returncode != 0:
|
||
|
|
return f"Error: {result.stderr.strip()}"
|
||
|
|
return result.stdout.strip()
|
||
|
|
except Exception as e:
|
||
|
|
return f"Error: {str(e)}"
|
||
|
|
|
||
|
|
def aha_list_features(self, product: str, __user__: dict = {}) -> str:
|
||
|
|
"""
|
||
|
|
List features for an Aha! product.
|
||
|
|
|
||
|
|
:param product: Aha! product key
|
||
|
|
:return: JSON list of features
|
||
|
|
"""
|
||
|
|
env = os.environ.copy()
|
||
|
|
env["AHA_DOMAIN"] = self.valves.AHA_DOMAIN
|
||
|
|
env["AHA_API_TOKEN"] = self.valves.AHA_API_KEY
|
||
|
|
|
||
|
|
try:
|
||
|
|
result = subprocess.run(
|
||
|
|
[
|
||
|
|
"mcporter",
|
||
|
|
"--config", self.valves.MCPORTER_CONFIG,
|
||
|
|
"call", "aha", "list_features",
|
||
|
|
"--params", json.dumps({"product": product})
|
||
|
|
],
|
||
|
|
capture_output=True, text=True, timeout=30, env=env
|
||
|
|
)
|
||
|
|
if result.returncode != 0:
|
||
|
|
return f"Error: {result.stderr.strip()}"
|
||
|
|
return result.stdout.strip()
|
||
|
|
except Exception as e:
|
||
|
|
return f"Error: {str(e)}"
|