""" Task Generation Agent for Task Manager MCP Servers Adapted from the original project's TaskGenerationAgent for use in MCP servers. This agent converts requirements documents into structured task lists. """ import os import json import requests import logging from typing import Dict, List, Optional logger = logging.getLogger(__name__) class TaskGenerationAgent: """ 任务生成Agent - 负责将用户的需求解析成高阶的可执行任务列表。 """ SYSTEM_PROMPT_TEMPLATE = """ You are a Chief Product Officer (CPO) AI. Your sole responsibility is to break down a user's project request into 3 to 5 high-level, strategic "Epics" or "Features". **CRITICAL RULES:** 1. **Think Strategically:** Do not think about small developer tasks. Your job is to define the core, high-level functional blocks of the project. 2. **Strict Quantity Control:** You MUST break the project into exactly 3 to 5 core Epics. Do NOT create more than 5. 3. **Output Format:** Your response MUST be a raw JSON object, starting with `{` and ending with `}`. It must contain a single root key "tasks", which is a list of task objects. 4. **Task Object Schema:** Each task object must have: `id` (string), `name` (string, the Epic's title), `description` (string, what this Epic is about and its goal), and `is_complex` (boolean, which must always be `true`). --- **EXAMPLE** *Input:* ``` # Project Requirements: Simple Todo App ## 1. Vision & Goals Create a basic todo application for personal task management. ## 2. Core Features - Task creation and editing - Task completion tracking - Simple categorization - Basic search functionality ## 3. Target Users Individual users who need simple task management. ## 4. Technical Constraints - Web-based application - Simple and intuitive interface ``` *Output:* ```json { "tasks": [ { "id": "epic-001", "name": "User Interface Design & Implementation", "description": "Design and implement the core user interface for task management including task creation, editing, and display components.", "is_complex": true }, { "id": "epic-002", "name": "Task Management Core Logic", "description": "Implement the core business logic for task operations including CRUD operations, status management, and data persistence.", "is_complex": true }, { "id": "epic-003", "name": "Search and Categorization Features", "description": "Build search functionality and categorization system to help users organize and find their tasks efficiently.", "is_complex": true } ] } ``` --- """ def __init__(self, api_key: Optional[str] = None, base_url: Optional[str] = None): self.api_key = api_key or os.environ.get("OPENAI_API_KEY") self.base_url = base_url or os.environ.get("OPENAI_BASE_URL", "https://api.openai.com/v1") if not self.api_key: raise ValueError("API key must be provided or set as environment variable (OPENAI_API_KEY).") def parse_document_to_tasks(self, document: str) -> List[Dict]: """ Parse a requirements document into a structured list of high-level tasks. """ logger.info("--- Task Generation Agent ---") logger.info("⏳ Calling LLM to parse requirements into a JSON task list...") api_url = f"{self.base_url}/chat/completions" if not self.base_url.endswith('/v1'): api_url = f"{self.base_url}/v1/chat/completions" headers = { "Content-Type": "application/json", "Authorization": f"Bearer {self.api_key}" } # Ensure proper encoding for the payload try: encoded_system_prompt = self.SYSTEM_PROMPT_TEMPLATE.encode('utf-8', errors='replace').decode('utf-8') encoded_document = document.encode('utf-8', errors='replace').decode('utf-8') except (UnicodeEncodeError, UnicodeDecodeError): encoded_system_prompt = self.SYSTEM_PROMPT_TEMPLATE encoded_document = document payload = { "model": "gpt-4o", "messages": [ {"role": "system", "content": encoded_system_prompt}, {"role": "user", "content": encoded_document} ], "temperature": 0.05, } try: # Add retry mechanism for network issues import time max_retries = 3 for attempt in range(max_retries): try: # Use json parameter instead of data for proper encoding response = requests.post( api_url, headers=headers, json=payload, # This automatically handles encoding timeout=60, # Increased timeout verify=True # Ensure SSL verification ) response.raise_for_status() break except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as e: if attempt < max_retries - 1: print(f"Network error on attempt {attempt + 1}, retrying in 2 seconds...") time.sleep(2) continue else: raise Exception(f"Network connection failed after {max_retries} attempts: {str(e)}") response_data = response.json() content = response_data.get("choices", [{}])[0].get("message", {}).get("content", "") # Ensure proper encoding for the response try: content = content.encode('utf-8', errors='replace').decode('utf-8') except (UnicodeEncodeError, UnicodeDecodeError): pass if not content: logger.warning("LLM returned empty content.") return [] # Try to parse JSON directly try: result = json.loads(content) tasks = result.get("tasks", []) logger.info(f"✅ Successfully parsed {len(tasks)} high-level tasks.") return tasks except json.JSONDecodeError: # Try to extract JSON from the content import re json_match = re.search(r'\{.*\}', content, re.DOTALL) if json_match: try: result = json.loads(json_match.group()) tasks = result.get("tasks", []) logger.info(f"✅ Extracted and parsed {len(tasks)} high-level tasks from response.") return tasks except json.JSONDecodeError: pass logger.error(f"Failed to parse LLM response as JSON: {content}") return [] except requests.exceptions.RequestException as e: logger.error(f"API call failed: {e}") return [] except Exception as e: logger.error(f"An unknown error occurred: {e}") return []