Clean up: Remove backup_v1 directory and update configuration files

backup_v1/agent/model_client.py

@@ -1,75 +0,0 @@
-import aiohttp
-import json
-from app.config import settings
-from app.prompts.system_prompts import topcoder_system_prompt
-
-class LLMClient:
-    def __init__(self):
-        self.api_url = settings.LLM_API_URL or "https://router.huggingface.co/v1/chat/completions"
-        self.headers = {
-            "Authorization": f"Bearer {settings.HF_TOKEN}",
-            "Content-Type": "application/json"
-        }
-        self.model = settings.HF_MODEL or "Qwen/Qwen2.5-7B-Instruct:together"
-        self.temperature = settings.LLM_TEMPERATURE or 0.8
-
-    async def chat(self, messages) -> str:
-        system_msg = {"role": "system", "content": topcoder_system_prompt}
-        payload = {
-            "model": self.model,
-            "messages": [system_msg] + messages,
-            "temperature": self.temperature
-        }
-
-        async with aiohttp.ClientSession() as session:
-            async with session.post(self.api_url, headers=self.headers, json=payload) as resp:
-                if resp.status != 200:
-                    text = await resp.text()
-                    raise Exception(f"LLM error {resp.status}: {text}")
-                data = await resp.json()
-                return data["choices"][0]["message"]["content"]
-
-    async def decide_tool(self, prompt: str) -> dict:
-        """Decide which tool to use based on dynamic prompt."""
-        
-        system_msg = {
-            "role": "system",
-            "content": "You are a tool decision assistant. Analyze the user's request and available tools/resources, then respond ONLY with a JSON object containing the tool name and parameters."
-        }
-        user_msg = {
-            "role": "user",
-            "content": prompt
-        }
-
-        response = await self.chat([system_msg, user_msg])
-
-        # Debug: Print the raw LLM response
-        print(f"🔍 LLM raw decision response:\n{response}")
-
-        # Try parsing the JSON
-        try:
-            return json.loads(response.strip())
-        except json.JSONDecodeError:
-            # Fallback: treat as chat if LLM fails
-            return {
-                "tool": "chat",
-                "params": {}
-            }
-            
-    async def complete_json(self, prompt: str) -> dict:
-        """
-        Calls the LLM to complete a JSON-only response based on a prompt.
-        """
-        raw_response = await self.chat([
-            {"role": "system", "content": "You are a strict API assistant. Only return valid JSON."},
-            {"role": "user", "content": prompt.strip()}
-        ])
-
-        try:
-            # Try to parse JSON safely
-            json_start = raw_response.find("{")
-            json_text = raw_response[json_start:]
-            return json.loads(json_text)
-        except Exception as e:
-            return {"error": f"Failed to parse JSON: {str(e)}", "raw": raw_response}
-

backup_v1/agent/prompt_engine.py

@@ -1,110 +0,0 @@
-from app.mcp.compact_utils import CompactSpecsUtils
-
-class PromptBuilder:
-    def __init__(self):
-        self.compact_utils = CompactSpecsUtils()
-    
-    def build_tool_decision_prompt(self, user_input: str) -> str:
-        """Build dynamic prompt based on available compact specs."""
-        
-        # Get available tools and resources
-        working_tools = self.compact_utils.get_working_tools()
-        working_resources = self.compact_utils.get_working_resources()
-        
-        # Build tools section with clear usage instructions
-        tools_section = ""
-        if working_tools:
-            tools_section = "Available tools:\n"
-            for tool_name in working_tools:
-                spec = self.compact_utils.get_tool_spec(tool_name)
-                if spec:
-                    description = spec.get("description", "")
-                    parameters = spec.get("parameters", {})
-                    
-                    # Create clear usage instructions with comprehensive parameter details
-                    if tool_name == "query-tc-skills":
-                        tools_section += f"- {tool_name}: Use this when the user asks about skills, technologies, or competencies.\n"
-                    elif tool_name == "query-tc-challenges":
-                        tools_section += f"- {tool_name}: Use this when the user asks about challenges, competitions, or contests.\n"
-                    else:
-                        tools_section += f"- {tool_name}: {description}.\n"
-                    
-                    # Add detailed parameter information
-                    if parameters:
-                        tools_section += f"  Parameters for {tool_name}:\n"
-                        for param_name, param_desc in parameters.items():
-                            # Truncate very long descriptions for readability
-                            short_desc = param_desc[:150] + "..." if len(param_desc) > 150 else param_desc
-                            tools_section += f"    - {param_name}: {short_desc}\n"
-        
-        # Build resources section
-        resources_section = ""
-        if working_resources:
-            resources_section = "Available API resources:\n"
-            for resource_name in working_resources:
-                spec = self.compact_utils.get_resource_spec(resource_name)
-                if spec:
-                    description = spec.get("description", "")
-                    total_endpoints = spec.get("total_endpoints", 0)
-                    key_endpoints = spec.get("key_endpoints", [])
-                    
-                    # Create clear usage instructions
-                    if "challenge" in resource_name.lower():
-                        resources_section += f"- {resource_name}: Use this for challenge API calls. Use /challenges endpoint.\n"
-                    elif "member" in resource_name.lower():
-                        resources_section += f"- {resource_name}: Use this for member API calls. Use /members endpoint.\n"
-                    else:
-                        resources_section += f"- {resource_name}: {description[:100]}...\n"
-                    
-                    # Add detailed parameter information from key endpoints
-                    if key_endpoints:
-                        for endpoint in key_endpoints[:1]:  # Show first endpoint parameters
-                            endpoint_params = endpoint.get("parameters", {})
-                            if endpoint_params:
-                                resources_section += f"  Parameters for {endpoint.get('path', '')}:\n"
-                                for param_name, param_desc in endpoint_params.items():
-                                    # Truncate very long descriptions for readability
-                                    short_desc = param_desc[:150] + "..." if len(param_desc) > 150 else param_desc
-                                    resources_section += f"    - {param_name}: {short_desc}\n"
-        
-        prompt = f"""
-You are an assistant that ONLY works with Topcoder's Member Communication Platform (MCP).
-
-Your goal is to decide the correct tool or resource to call, based ONLY on the user input.
-
-{tools_section}
-{resources_section}
-- chat: Use for casual greetings, thanks, or general questions that don't need any tool.
-- reject: Use if the query is not related to Topcoder.
-
-IMPORTANT: 
-- For tools: Only include parameters that have meaningful values from the user query. Do not include empty or default parameters.
-- For resources: Use the resource name and specify the endpoint path and method.
-- For member queries: Use /members with parameters like handle, page, perPage. For specific members, use /members/{{handle}}.
-- For challenge queries: Use /challenges with parameters like status, track, page, perPage.
-- Always include relevant parameters in the params object for API calls.
-
-Respond ONLY with a JSON object in this format:
-{{"tool": "<tool_name>", "params": {{"param1": "value1", "param2": "value2"}} }}
-OR
-{{"resource": "<resource_name>", "endpoint": "/path", "method": "GET", "params": {{"param1": "value1", "param2": "value2"}} }}
-
-User input: "{user_input}"
-"""
-        return prompt
-    
-    def get_available_tools(self) -> list:
-        """Get list of available tools for reference."""
-        return self.compact_utils.get_working_tools()
-    
-    def get_available_resources(self) -> list:
-        """Get list of available resources for reference."""
-        return self.compact_utils.get_working_resources()
-    
-    def get_tool_parameters(self, tool_name: str) -> dict:
-        """Get available parameters for a specific tool."""
-        return self.compact_utils.get_tool_parameters(tool_name)
-    
-    def get_resource_endpoints(self, resource_name: str) -> list:
-        """Get available endpoints for a specific resource."""
-        return self.compact_utils.get_resource_endpoints(resource_name)

backup_v1/agent/schema.py

@@ -1,29 +0,0 @@
-from typing import Optional, Literal, Dict, Any, Union
-from dataclasses import dataclass
-
-# Dynamic tool types from compact specs
-ToolType = str  # Will be populated from compact specs
-
-@dataclass
-class ToolRequest:
-    tool: ToolType
-    params: Optional[Dict[str, Any]] = None
-
-@dataclass
-class ToolResponse:
-    status: Literal["success", "error"]
-    data: Optional[Any] = None
-    message: Optional[str] = None
-
-@dataclass
-class CompactToolSpec:
-    """Compact tool specification from MCP catalog."""
-    description: str
-    parameters: Dict[str, str]
-
-@dataclass
-class CompactResourceSpec:
-    """Compact resource specification from MCP catalog."""
-    description: str
-    total_endpoints: int
-    key_endpoints: list

backup_v1/agent/tool_executor.py

@@ -1,127 +0,0 @@
-import asyncio
-from app.mcp.stream_client import load_server_config, StreamableHttpMCPClient
-from app.mcp.compact_utils import CompactSpecsUtils
-from app.agent.schema import ToolRequest, ToolResponse
-
-class ToolExecutor:
-    def __init__(self):
-        self.compact_utils = CompactSpecsUtils()
-    
-    async def execute(self, request: ToolRequest) -> ToolResponse:
-        """Execute a tool or resource request using compact specs and MCP client."""
-        try:
-            tool_name = request.tool
-            params = request.params or {}
-            
-            # Check if it's a tool
-            tool_spec = self.compact_utils.get_tool_spec(tool_name)
-            if tool_spec:
-                return await self._execute_tool(tool_name, params)
-            
-            # Check if it's a resource
-            resource_spec = self.compact_utils.get_resource_spec(tool_name)
-            if resource_spec:
-                return await self._execute_resource(tool_name, params)
-            
-            return ToolResponse(
-                status="error", 
-                message=f"'{tool_name}' not found in available tools or resources"
-            )
-                
-        except Exception as e:
-            return ToolResponse(status="error", message=str(e))
-    
-    async def _execute_tool(self, tool_name: str, params: dict) -> ToolResponse:
-        """Execute a tool request."""
-        try:
-            # Format the tool request using compact specs
-            formatted_request = self.compact_utils.format_tool_request(tool_name, params)
-            
-            # Execute via MCP client
-            server_config = load_server_config()
-            async with StreamableHttpMCPClient(server_config) as client:
-                await client.initialize()
-                
-                # Make the tool call
-                result = await client.call("tools/call", formatted_request["params"])
-                
-                # Check for errors
-                if "error" in result:
-                    error_msg = result["error"].get("message", "Unknown error")
-                    return ToolResponse(status="error", message=error_msg)
-                
-                # Return success with data
-                return ToolResponse(status="success", data=result.get("result", result))
-                
-        except Exception as e:
-            return ToolResponse(status="error", message=str(e))
-    
-    async def _execute_resource(self, resource_name: str, params: dict) -> ToolResponse:
-        """Execute a resource (API) request using direct API calls."""
-        try:
-            # Extract endpoint and method from params
-            endpoint = params.get("endpoint", "/")
-            method = params.get("method", "GET")
-            api_params = {k: v for k, v in params.items() if k not in ["endpoint", "method"]}
-            
-            # Substitute path parameters in the endpoint
-            substituted_endpoint = endpoint
-            params_to_remove = []
-            for param_name, param_value in api_params.items():
-                placeholder = f"{{{param_name}}}"
-                if placeholder in substituted_endpoint:
-                    substituted_endpoint = substituted_endpoint.replace(placeholder, str(param_value))
-                    # Mark parameter for removal since it's now in the path
-                    params_to_remove.append(param_name)
-            
-            # Remove substituted parameters from api_params
-            for param_name in params_to_remove:
-                api_params.pop(param_name, None)
-            
-            # Make direct API call using compact specs
-            result = await self.compact_utils.make_direct_api_call(
-                resource_name, substituted_endpoint, method, api_params
-            )
-            
-            # Check for errors
-            if result.get("status") != 200:
-                error_msg = result.get("error", "Unknown error")
-                return ToolResponse(status="error", message=f"API error {result.get('status')}: {error_msg}")
-            
-            # Return success with data
-            return ToolResponse(status="success", data=result.get("data", result))
-                
-        except Exception as e:
-            return ToolResponse(status="error", message=str(e))
-    
-    def get_available_tools(self) -> list:
-        """Get list of available tools."""
-        return self.compact_utils.get_working_tools()
-    
-    def get_available_resources(self) -> list:
-        """Get list of available resources."""
-        return self.compact_utils.get_working_resources()
-    
-    def get_tool_parameters(self, tool_name: str) -> dict:
-        """Get available parameters for a tool."""
-        return self.compact_utils.get_tool_parameters(tool_name)
-    
-    def get_resource_endpoints(self, resource_name: str) -> list:
-        """Get available endpoints for a resource."""
-        return self.compact_utils.get_resource_endpoints(resource_name)
-    
-    def validate_tool_request(self, tool_name: str, params: dict) -> bool:
-        """Validate if a tool request is valid."""
-        try:
-            self.compact_utils.format_tool_request(tool_name, params)
-            return True
-        except Exception:
-            return False
-    
-    def validate_resource_request(self, resource_name: str, endpoint: str, method: str = "GET") -> bool:
-        """Validate if a resource request is valid."""
-        try:
-            self.compact_utils.format_resource_request(resource_name, endpoint, method)
-            return True
-        except Exception:
-            return False

backup_v1/config/api_endpoints.py

@@ -1,16 +0,0 @@
-# Hardcoded base URLs for Topcoder APIs
-# This file contains only the base URLs, parameters are fetched dynamically
-
-API_BASE_URLS = {
-    "Member_V6_API_Swagger": "https://api.topcoder.com/v5",
-    "Challenges_V6_API_Swagger": "https://api.topcoder.com/v5",
-    "Identity_V6_API_Swagger": "https://api.topcoder.com/v5"
-}
-
-def get_base_url(resource_name: str) -> str:
-    """Get base URL for a resource."""
-    return API_BASE_URLS.get(resource_name, "")
-
-def get_all_base_urls() -> dict:
-    """Get all available base URLs."""
-    return API_BASE_URLS.copy() 

backup_v1/config/compact_api_specs.py

@@ -1,65 +0,0 @@
-# Generated compact API specs from MCP catalog
-# Only includes working tools and resources
-
-COMPACT_API_SPECS = {
-    # Working Tools
-    "query-tc-skills": {
-        "description": "Returns a list of standardized skills from Topcoder platform, filtered and sorted based on the provided parameters.",
-        "parameters": {
-            "name": "Filter by skill names, exact match. Array of string. Optional",
-            "skillId": "Filter by skill IDs, exact match. Array of string. Optional",
-            "sortBy": "Sort skills by a specific field. Type: string. Optional. Valid values: \"name\", \"description\", \"created_at\", \"updated_at\"",
-            "sortOrder": "Sort order. Type: string. Optional. Valid values: \"ASC\", \"DESC\"",
-            "page": "Page number for pagination, starting from 1. Type: number. Optional. Default: 1",
-            "perPage": "Number of standardized skills per page, between 1 and 100. Type: number. Optional. Default: 20",
-        }
-    },
-    "query-tc-challenges": {
-        "description": "Returns a list of Topcoder challenges based on the query parameters.",
-        "parameters": {
-            "id": "Filter by challenge ID, exact match. Type: string. Optional",
-            "status": "Filter by challenge status. Type: string. Optional. Valid values: \"New\", \"Draft\", \"Cancelled\", \"Active\", \"Completed\", \"Deleted\", \"Cancelled - Failed Review\", \"Cancelled - Failed Screening\"... (and 5 more)",
-            "type": "Filter by type abbreviation, exact match. Type: string. Optional",
-            "track": "Filter by track, case-insensitive, partial matches are allowed. Type: string. Optional",
-            "tag": "Filter by tag name, case-insensitive, partial matches are allowed. Type: string. Optional",
-            "tags": "Filter by multiple tag names, case-insensitive, partial matches are allowed. Array of string. Optional",
-            "search": "Filter by name, description and tags fields, case-insensitive, partial matches are allowed. Type: string. Optional",
-            "startDateStart": "Filter by start date (lower bound of date range, ISO format). Type: string. Optional",
-            "startDateEnd": "Filter by start date (upper bound of date range, ISO format). Type: string. Optional",
-            "currentPhaseName": "Filter by current phase name. Type: string. Optional",
-            "sortBy": "Sort challenges by a specific field. Type: string. Optional. Valid values: \"updatedBy\", \"updated\", \"createdBy\", \"created\", \"endDate\", \"startDate\", \"projectId\", \"name\"... (and 5 more)",
-            "sortOrder": "Sort order. Type: string. Optional. Valid values: \"asc\", \"desc\"",
-            "totalPrizesFrom": "Filter by the lowest amount of total prizes on the challenge. Type: number. Optional",
-            "totalPrizesTo": "Filter by the highest amount of total prizes on the challenge. Type: number. Optional",
-            "createdBy": "Filter by the user who created the challenge. Type: string. Optional",
-            "page": "Page number for pagination, starting from 1. Type: number. Optional. Default: 1",
-            "perPage": "Number of challenges per page, between 1 and 100. Type: number. Optional. Default: 20",
-        }
-    },
-
-    # Working Resources
-    "Challenges_V6_API_Swagger": {
-        "description": "Challenge V5 API: ## Pagination - Requests that return multiple items will be paginated to 20 items by default. - You can specify further pages with the `page` paramete...",
-        "total_endpoints": 11,
-        "base_url": "https://api.topcoder.com/v5",
-        "key_endpoints": [
-            {"path": "/challenges", "method": "GET", "full_url": "https://api.topcoder.com/v5/challenges", "description": "Available parameters: id, selfService, selfServiceCopilot, confidentialityType, directProjectId, typeIds, trackIds, types, tracks, typeId, trackId, type, track, name, search, description, timelineTemplateId, reviewType, tag, tags, includeAllTags, projectId, forumId, legacyId, status, group, startDateStart, startDateEnd, endDateStart, endDateEnd, currentPhaseName, createdDateStart, createdDateEnd, updatedDateStart, updatedDateEnd, registrationStartDateStart, registrationStartDateEnd, registrationEndDateStart, registrationEndDateEnd, submissionStartDateStart, submissionStartDateEnd, submissionEndDateStart, submissionEndDateEnd, createdBy, updatedBy, isLightweight, memberId, sortBy, sortOrder, isTask, taskIsAssigned, taskMemberId, events, tco, includeAllEvents, useSchedulingAPI, totalPrizesFrom, totalPrizesTo", "parameters": {"id": "Filter by id, exact match. Type: string. Location: query. Optional. Format: UUI", "selfService": "Filter by selfService flag. Type: boolean. Location: query. Optional", "selfServiceCopilot": "Filter by selfServiceCopilot. Type: string. Location: query. Optional", "confidentialityType": "Filter by confidentialityType. Type: string. Location: query. Optional. Valid values: \\\"public\\\", \\\"private\\\"", "directProjectId": "Filter by directProjectId. Type: integer. Location: query. Optional", "typeIds": "Filter by multiple type IDs, exact match.. Type: array. Location: query. Optional", "trackIds": "Filter by multiple track IDs, exact match.. Type: array. Location: query. Optional", "types": "Filter by multiple type abbreviation, exact match. If types is provided, typeIds will be ignored. Type: array. Location: query. Optional", "tracks": "Filter by multiple track abbreviation, exact match. If tracks is provided, trackIds will be ignored. Type: array. Location: query. Optional", "typeId": "Filter by type id, exact match. If type is provided, typeId will be ignored. Type: string. Location: query. Optional. Format: UUID", "trackId": "Filter by track id, exact match. If track is provided, trackId will be ignored. Type: string. Location: query. Optional. Format: UUID", "type": "Filter by type abbreviation, exact match. If provided, the typeId will be ignored. Type: string. Location: query. Optional", "track": "Filter by track, case-insensitive, partial matches are allowed.. Type: string. Location: query. Optional", "name": "Filter by name, case-insensitive, partial matches are allowed. If search is provided, name will be ignored. Type: string. Location: query. Optional", "search": "Filter by name, description and tags fields, case-insensitive, partial matches are allowed. If search is provided, name and description will be ignored.. Type: string. Location: query. Optional", "description": "Filter by description, case-insensitive, partial matches are allowed. If search is provided, description will be ignored.. Type: string. Location: query. Optional", "timelineTemplateId": "Filter by timeline template id, exact match. Type: string. Location: query. Optional", "reviewType": "Filter by review type, case-insensitive, partial matches are allowed.. Type: string. Location: query. Optional", "tag": "Filter by tag name, case-insensitive, partial matches are allowed.. Type: string. Location: query. Optional", "tags": "Filter by multiple tag names, case-insensitive, partial matches are allowed.. Type: array. Location: query. Optional", "includeAllTags": "Require all provided tags to be present on a challenge for a match.. Type: boolean. Location: query. Optional. Default: True", "projectId": "Filter by v5 project id, exact match.. Type: integer. Location: query. Optional", "forumId": "Filter by forum id, exact match.. Type: integer. Location: query. Optional", "legacyId": "Filter by legacy id, exact match.. Type: integer. Location: query. Optional", "status": "Filter by status, case-insensitive, exact match.. Type: string. Location: query. Optional. Valid values: \\\"New\\\", \\\"Draft\\\", \\\"Cancelled\\\", \\\"Active\\\", \\\"Completed\\\", \\\"Deleted\\\"... (and 7 more)", "group": "Filter by group name, case-insensitive, partial matches are allowed.. Type: string. Location: query. Optional", "startDateStart": "Filter by start date (lower bound of date range). Type: string. Location: query. Optional. Format: date-time", "startDateEnd": "Filter by start date (upper bound of date range). Type: string. Location: query. Optional. Format: date-time", "endDateStart": "Filter by end date (lower bound of date range). Type: string. Location: query. Optional. Format: date-time", "endDateEnd": "Filter by end date (upper bound of date range). Type: string. Location: query. Optional. Format: date-time", "currentPhaseName": "Filter by name of the current phase. Type: string. Location: query. Optional", "createdDateStart": "Filter by created date (lower bound of date range). Type: string. Location: query. Optional. Format: date-time", "createdDateEnd": "Filter by created date (upper bound of date range). Type: string. Location: query. Optional. Format: date-time", "updatedDateStart": "Filter by updated date (lower bound of date range). Type: string. Location: query. Optional. Format: date-time", "updatedDateEnd": "Filter by updated date (upper bound of date range). Type: string. Location: query. Optional. Format: date-time", "registrationStartDateStart": "Filter by registration start date (lower bound of date range). Type: string. Location: query. Optional. Format: date-time", "registrationStartDateEnd": "Filter by registration start date (upper bound of date range). Type: string. Location: query. Optional. Format: date-time", "registrationEndDateStart": "Filter by registration end date (lower bound of date range). Type: string. Location: query. Optional. Format: date-time", "registrationEndDateEnd": "Filter by registration end date (upper bound of date range). Type: string. Location: query. Optional. Format: date-time", "submissionStartDateStart": "Filter by submission start date (lower bound of date range). Type: string. Location: query. Optional. Format: date-time", "submissionStartDateEnd": "Filter by submission start date (upper bound of date range). Type: string. Location: query. Optional. Format: date-time", "submissionEndDateStart": "Filter by submission end date (lower bound of date range). Type: string. Location: query. Optional. Format: date-time", "submissionEndDateEnd": "Filter by submission end date (upper bound of date range). Type: string. Location: query. Optional. Format: date-time", "createdBy": "Filter by 'createdBy' field, case-insensitive, partial matches are allowed.. Type: string. Location: query. Optional", "updatedBy": "Filter by 'updatedBy' field, case-insensitive, partial matches are allowed.. Type: string. Location: query. Optional", "isLightweight": "Turn off description and privateDescription for a smaller response payload. Type: boolean. Location: query. Optional. Default: False", "memberId": "Filter challenges memberId has access to. Type: string. Location: query. Optional", "sortBy": "Sort the results by the field.. Type: string. Location: query. Optional. Valid values: \\\"updatedBy\\\", \\\"updated\\\", \\\"createdBy\\\", \\\"created\\\", \\\"endDate\\\", \\\"startDate\\\"... (and 7 more)", "sortOrder": "Order the results by the asc/desc.. Type: string. Location: query. Optional. Valid values: \\\"asc\\\", \\\"desc\\\"", "isTask": "Filter based on the task.isTask property. Only available for admins/m2m. Will be ignored for regular users or not authenticated users. Type: boolean. Location: query. Optional", "taskIsAssigned": "Filter based on the task.isAssigned property. Only available for admins/m2m. Will be ignored for regular users or not authenticated users. Type: boolean. Location: query. Optional", "taskMemberId": "Filter based on the task.memberId property. Only available for admins/m2m. Will be ignored for regular users or not authenticated users. Type: string. Location: query. Optional", "events": "Filter by multiple event keys (tco21). Type: array. Location: query. Optional", "tco": "Filter by tco eligible events. Type: boolean. Location: query. Optional", "includeAllEvents": "Require all provided events to be present on a challenge for a match. Type: boolean. Location: query. Optional. Default: True", "useSchedulingAPI": "Search based on 'legacy.useSchedulingAPI'. Type: boolean. Location: query. Optional", "totalPrizesFrom": "Filter by the lowest amount of total prizes on the challenge. Type: number. Location: query. Optional", "totalPrizesTo": "Filter by the highest amount of total prizes on the challenge. Type: number. Location: query. Optional"}},
-        ]
-    },
-    "Member_V6_API_Swagger": {
-        "description": "Topcoder Member API: Services that provide access and interaction with all sorts of member profile details.  # Pagination  - Requests that return multiple items will be pa...",
-        "total_endpoints": 10,
-        "base_url": "https://api.topcoder.com/v5",
-        "key_endpoints": [
-            {"path": "/members", "method": "GET", "full_url": "https://api.topcoder.com/v5/members", "description": "Available parameters: userId, userIds, handle, handles, handleLower, handlesLower, sort, fields", "parameters": {"userId": "filter by userId. Type: string. Location: query. Optional", "userIds": "filter by userIds. Example - [userId1,userId2,userId3,...,userIdN]. Type: string. Location: query. Optional", "handle": "filter by handle. This will return like search.. Type: string. Location: query. Optional", "handles": "filter by handles. This will return like search. Example - [\\\"handle1\\\",\\\"handle2\\\",\\\"handle3\\\",...,\\\"handleN\\\"]. Type: string. Location: query. Optional", "handleLower": "filter by handle. This will return an exact search.. Type: string. Location: query. Optional", "handlesLower": "filter by handles. This will return an exact search. Example - [\\\"handle1\\\",\\\"handle2\\\",\\\"handle3\\\",...,\\\"handleN\\\"]. Type: string. Location: query. Optional", "sort": "sort by asc or desc. Type: string. Location: query. Optional", "fields": "fields=fieldName1,fieldName2,...,fieldN  parameter for choosing which fields of members profile that will be included in response.. Type: string. Location: query. Optional"}},
-        ]
-    },
-    "Identity_V6_API_Swagger": {
-        "description": "TC Service Identity",
-        "total_endpoints": 47,
-        "base_url": "https://api.topcoder.com/v5",
-        "key_endpoints": [
-            {"path": "/groups/{groupId}/singleMember/{memberId}", "method": "GET", "full_url": "https://api.topcoder.com/v5/groups/{groupId}/singleMember/{memberId}"},
-        ]
-    },
-}

backup_v1/config/settings.py

@@ -1,27 +0,0 @@
-from dotenv import load_dotenv
-import os
-
-load_dotenv()
-
-# LLM Configuration
-HF_TOKEN = os.getenv("HF_TOKEN")
-HF_MODEL = os.getenv("HF_MODEL", "Qwen/Qwen2.5-7B-Instruct:together")
-LLM_API_URL = os.getenv("LLM_API_URL", "https://router.huggingface.co/v1/chat/completions")
-LLM_TEMPERATURE = float(os.getenv("LLM_TEMPERATURE", "0.8"))
-
-# MCP Configuration
-# Default to provided dev MCP endpoint; can be overridden via env
-MCP_BASE_URL = os.getenv("MCP_BASE_URL", "https://api.topcoder-dev.com/v6/mcp/mcp")
-MCP_SERVER_URL = os.getenv("MCP_SERVER_URL", "https://api.topcoder-dev.com/v6/mcp/mcp")
-MCP_TOKEN = os.getenv("MCP_TOKEN")
-
-# API Authentication Configuration
-# Set to True to exclude endpoints that require authentication
-EXCLUDE_AUTH_REQUIRED_ENDPOINTS = os.getenv("EXCLUDE_AUTH_REQUIRED_ENDPOINTS", "true").lower() == "true"
-
-# Additional authentication filtering options
-# Set to True to also exclude endpoints that have 401 responses in their documentation
-EXCLUDE_401_RESPONSES = os.getenv("EXCLUDE_401_RESPONSES", "true").lower() == "true"
-
-# Set to True to be more strict and exclude endpoints that mention authentication in descriptions
-STRICT_AUTH_FILTERING = os.getenv("STRICT_AUTH_FILTERING", "false").lower() == "true"

backup_v1/config/static_responses.py

@@ -1,6 +0,0 @@
-# config/static_responses.py
-
-REJECT_RESPONSE = (
-    "I'm here to help with Topcoder-related queries only. "
-    "Please ask something related to Topcoder challenges, members, or platform features."
-)

backup_v1/main.py

@@ -1,133 +0,0 @@
-# main.py
-
-import asyncio
-
-from app.agent.model_client import LLMClient
-from app.ui.interface import launch_ui
-
-
-
-# Optional: Manual test for LLM
-async def test_llm():
-    client = LLMClient()
-    result = await client.chat([
-        {"role": "user", "content": "What is Topcoder?"}
-    ])
-    print("LLM Response:\n", result)
-
-
-# Optional: Manual test for MCP API
-async def test_mcp():
-    client = MCPClient()
-    challenges = await client.get_challenges()
-    print("Sample Challenges:\n", challenges[:2])
-
-    member = await client.get_member("tourist")
-    print("Member Info:\n", member)
-
-
-async def initialize_mcp_registry():
-    """Connect to MCP server, list tools/resources, and store them in registry."""
-    from app.mcp.stream_client import load_server_config, StreamableHttpMCPClient
-    from app.mcp.catalog import MCPCatalog
-    from app.mcp.analyzers import OpenAPIAnalyzer
-    from app.mcp.specs_generator import SpecsGenerator
-    
-    try:
-        server_config = load_server_config()
-        catalog = MCPCatalog()
-        analyzer = OpenAPIAnalyzer()
-        
-        async with StreamableHttpMCPClient(server_config) as client:
-            # Initialize the session
-            init_data = await client.initialize()
-            print("✅ MCP session initialized")
-            
-            # Send initialized notification
-            try:
-                status, body = await client.send_initialized_notification()
-                print(f"✅ Initialized notification sent (HTTP {status})")
-            except Exception as e:
-                print(f"⚠️ Failed to send initialized notification: {e}")
-            
-            # List and catalog tools
-            print("\n🔧 Cataloging tools...")
-            tools_data = await client.call("tools/list", {})
-            tools = tools_data.get("result", {}).get("tools", [])
-            
-            for tool in tools:
-                tool_name = tool.get("name", "unknown")
-                print(f"  📝 Cataloging tool: {tool_name}")
-                try:
-                    catalog.save_tool_schema(tool_name, tool)
-                    print(f"    ✅ Saved schema for {tool_name}")
-                except Exception as e:
-                    print(f"    ❌ Failed to save schema for {tool_name}: {e}")
-            
-            # List and catalog resources
-            print("\n📚 Cataloging resources...")
-            resources_data = await client.call("resources/list", {})
-            resources = resources_data.get("result", {}).get("resources", [])
-            
-            for resource in resources:
-                resource_name = resource.get("name", "unknown")
-                resource_uri = resource.get("uri", "")
-                print(f"  📝 Cataloging resource: {resource_name}")
-                
-                try:
-                    # Analyze API resources (Swagger/OpenAPI docs)
-                    if "swagger" in resource_uri.lower() or "api-docs" in resource_uri.lower():
-                        print(f"    🔍 Analyzing API documentation: {resource_uri}")
-                        analysis = await analyzer.fetch_and_analyze_api(resource_uri)
-                        catalog.save_resource_analysis(resource_name, analysis)
-                        print(f"    ✅ Analyzed and saved {resource_name}")
-                    else:
-                        # For non-API resources, just save the basic info
-                        basic_analysis = {
-                            "name": resource_name,
-                            "uri": resource_uri,
-                            "description": resource.get("description", ""),
-                            "summary": f"Resource: {resource_name}",
-                            "url": resource_uri
-                        }
-                        catalog.save_resource_analysis(resource_name, basic_analysis)
-                        print(f"    ✅ Saved basic info for {resource_name}")
-                        
-                except Exception as e:
-                    print(f"    ❌ Failed to catalog {resource_name}: {e}")
-            
-            print(f"\n✅ Loaded {len(resources)} resources and {len(tools)} tools from MCP server")
-            
-        # Print catalog summary
-        summary = catalog.get_catalog_summary()
-        print(f"📁 Catalog stored in: {summary['catalog_dir']}")
-        print(f"📋 Cataloged tools: {len(summary['tools'])}")
-        print(f"📋 Cataloged resources: {len(summary['resources'])}")
-        
-        # Generate compact specs
-        print("\n🔧 Generating compact API specs...")
-        specs_generator = SpecsGenerator()
-        compact_specs = await specs_generator.generate_compact_specs()
-        specs_generator.save_compact_specs(compact_specs)
-        specs_generator.print_summary(compact_specs)
-            
-    except Exception as e:
-        print(f"⚠️ Failed to initialize MCP registry: {e}")
-        import traceback
-        traceback.print_exc()
-    finally:
-        await analyzer.close()
-
-
-# Note: Cleanup is handled automatically by StreamableHttpMCPClient context managers
-
-
-if __name__ == "__main__":
-    # Uncomment the tests below if you want to run them manually
-
-    # asyncio.run(test_llm())
-    # asyncio.run(test_mcp())
-
-    # Initialize MCP tool/resource registry, then launch UI
-    asyncio.run(initialize_mcp_registry())
-    launch_ui()

backup_v1/mcp-config.json

@@ -1,8 +0,0 @@
-{
-  "mcpServers": {
-    "topcoder": {
-      "url": "https://api.topcoder-dev.com/v6/mcp/mcp",
-      "description": "Topcoder MCP Server for challenges and skills"
-    }
-  }
-} 

backup_v1/mcp/analyzers.py

@@ -1,185 +0,0 @@
-import json
-import yaml
-import aiohttp
-from typing import Dict, List, Any, Optional
-from urllib.parse import urlparse
-
-
-class OpenAPIAnalyzer:
-    """Analyzes OpenAPI/Swagger documents to extract relevant API structures."""
-    
-    def __init__(self):
-        self.session: Optional[aiohttp.ClientSession] = None
-    
-    async def _get_session(self) -> aiohttp.ClientSession:
-        if not self.session:
-            self.session = aiohttp.ClientSession()
-        return self.session
-    
-    async def close(self):
-        if self.session:
-            await self.session.close()
-            self.session = None
-    
-    async def fetch_and_analyze_api(self, url: str) -> Dict[str, Any]:
-        """Fetch and analyze an OpenAPI/Swagger document from URL."""
-        session = await self._get_session()
-        
-        try:
-            async with session.get(url) as resp:
-                if resp.status != 200:
-                    raise Exception(f"Failed to fetch API doc: HTTP {resp.status}")
-                
-                content_type = resp.headers.get("content-type", "")
-                text = await resp.text()
-                
-                # Parse based on content type
-                if "yaml" in content_type or "yml" in content_type or url.endswith(('.yaml', '.yml')):
-                    doc = yaml.safe_load(text)
-                else:
-                    doc = json.loads(text)
-                
-                return self._analyze_openapi_doc(doc, url)
-                
-        except Exception as e:
-            return {
-                "error": f"Failed to analyze API at {url}: {str(e)}",
-                "url": url,
-                "summary": "API documentation could not be analyzed"
-            }
-    
-    def _analyze_openapi_doc(self, doc: Dict[str, Any], url: str) -> Dict[str, Any]:
-        """Analyze OpenAPI document and extract relevant information."""
-        info = doc.get("info", {})
-        paths = doc.get("paths", {})
-        servers = doc.get("servers", [])
-        
-        # Extract base URL from servers or use known Topcoder API URLs
-        base_url = None
-        if servers:
-            # Use the first server URL as base
-            base_url = servers[0].get("url", "")
-            # If it's a relative URL, combine with the original URL
-            if base_url.startswith("/"):
-                parsed_url = urlparse(url)
-                base_url = f"{parsed_url.scheme}://{parsed_url.netloc}{base_url}"
-        else:
-            # Use known Topcoder API URLs based on the document URL
-            if "member-api" in url:
-                base_url = "https://api.topcoder.com/v5"
-            elif "challenge-api" in url:
-                base_url = "https://api.topcoder.com/v5"
-            elif "identity-api" in url:
-                base_url = "https://api.topcoder.com/v5"
-            else:
-                # Try to extract from the URL itself
-                parsed_url = urlparse(url)
-                if "github.com" in parsed_url.netloc:
-                    # For GitHub raw URLs, use the standard Topcoder API
-                    base_url = "https://api.topcoder.com/v5"
-        
-        # Extract version from info (but don't add to base URL for Topcoder APIs)
-        version = info.get("version", "v5")
-        
-        # For Topcoder APIs, use the standard format without additional version suffixes
-        full_base_url = base_url
-        # Don't add version to base URL for Topcoder APIs since they use /v5/ format
-        
-        # Extract basic info
-        analysis = {
-            "url": url,
-            "base_url": full_base_url,
-            "version": version,
-            "title": info.get("title", "Unknown API"),
-            "description": info.get("description", ""),
-            "summary": "",
-            "endpoints": [],
-            "schemas": {}
-        }
-        
-        # Analyze endpoints
-        for path, path_item in paths.items():
-            for method, operation in path_item.items():
-                if method.lower() in ["get", "post", "put", "delete", "patch"]:
-                    # Construct full URL with base URL and version
-                    full_url = None
-                    if full_base_url:
-                        # Ensure path starts with /
-                        clean_path = path if path.startswith("/") else f"/{path}"
-                        full_url = f"{full_base_url}{clean_path}"
-                    
-                    # Check if endpoint requires authentication (has 401 response)
-                    requires_auth = "401" in operation.get("responses", {})
-                    
-                    # Additional checks for authentication requirements
-                    description = operation.get("description", "").lower()
-                    summary = operation.get("summary", "").lower()
-                    
-                    # Check for authentication mentions in description/summary
-                    auth_keywords = ["authentication", "authorization", "bearer", "token", "auth", "login", "credential"]
-                    has_auth_mention = any(keyword in description or keyword in summary for keyword in auth_keywords)
-                    
-                    # Check for secured field mentions
-                    secured_mentions = ["secured", "private", "admin", "authenticated", "authorized"]
-                    has_secured_mention = any(keyword in description or keyword in summary for keyword in secured_mentions)
-                    
-                    # Final authentication requirement
-                    requires_auth = requires_auth or has_auth_mention or has_secured_mention
-                    
-                    endpoint = {
-                        "path": path,
-                        "method": method.upper(),
-                        "summary": operation.get("summary", ""),
-                        "description": operation.get("description", ""),
-                        "parameters": operation.get("parameters", []),
-                        "requestBody": operation.get("requestBody", {}),
-                        "responses": operation.get("responses", {}),
-                        "full_url": full_url,
-                        "requires_auth": requires_auth
-                    }
-                    analysis["endpoints"].append(endpoint)
-        
-        # Extract schemas if available
-        if "components" in doc and "schemas" in doc["components"]:
-            schemas = doc["components"]["schemas"]
-            for schema_name, schema in schemas.items():
-                if isinstance(schema, dict):
-                    analysis["schemas"][schema_name] = {
-                        "type": schema.get("type", "object"),
-                        "properties": schema.get("properties", {}),
-                        "required": schema.get("required", []),
-                        "description": schema.get("description", "")
-                    }
-        
-        # Generate summary
-        analysis["summary"] = f"API with {len(analysis['endpoints'])} endpoints and {len(analysis['schemas'])} schemas"
-        
-        return analysis
-    
-    def _extract_relevant_schemas(self, analysis: Dict[str, Any]) -> List[Dict[str, Any]]:
-        """Extract the most relevant schemas from the analysis."""
-        schemas = []
-        
-        for schema_name, schema in analysis.get("schemas", {}).items():
-            # Focus on schemas that are likely to be important
-            if schema.get("type") == "object" and schema.get("properties"):
-                relevant_schema = {
-                    "name": schema_name,
-                    "type": schema.get("type"),
-                    "description": schema.get("description", ""),
-                    "properties": {}
-                }
-                
-                # Extract key properties (limit to first 10 to avoid overwhelming)
-                properties = schema.get("properties", {})
-                for prop_name, prop_info in list(properties.items())[:10]:
-                    if isinstance(prop_info, dict):
-                        relevant_schema["properties"][prop_name] = {
-                            "type": prop_info.get("type", "unknown"),
-                            "description": prop_info.get("description", ""),
-                            "required": prop_name in schema.get("required", [])
-                        }
-                
-                schemas.append(relevant_schema)
-        
-        return schemas 

backup_v1/mcp/catalog.py

@@ -1,114 +0,0 @@
-import json
-import os
-from typing import Dict, List, Any, Optional
-from pathlib import Path
-
-
-class MCPCatalog:
-    """Manages structured storage of MCP tools and resources."""
-    
-    def __init__(self, catalog_dir: str = "mcp_catalog"):
-        self.catalog_dir = Path(catalog_dir)
-        self.catalog_dir.mkdir(exist_ok=True)
-        
-        # Ensure subdirectories exist
-        (self.catalog_dir / "tools").mkdir(exist_ok=True)
-        (self.catalog_dir / "resources").mkdir(exist_ok=True)
-    
-    def save_tool_schema(self, tool_name: str, tool_data: Dict[str, Any]) -> str:
-        """Save tool schema to disk."""
-        file_path = self.catalog_dir / "tools" / f"{tool_name}.json"
-        
-        # Extract relevant tool information
-        schema = {
-            "name": tool_name,
-            "description": tool_data.get("description", ""),
-            "inputSchema": tool_data.get("inputSchema", {}),
-            "outputSchema": tool_data.get("outputSchema", {}),
-            "parameters": tool_data.get("parameters", []),
-            "raw_data": tool_data  # Keep original data for reference
-        }
-        
-        with open(file_path, 'w', encoding='utf-8') as f:
-            json.dump(schema, f, indent=2, ensure_ascii=False)
-        
-        return str(file_path)
-    
-    def save_resource_analysis(self, resource_name: str, analysis: Dict[str, Any]) -> str:
-        """Save resource analysis to disk."""
-        # Sanitize filename
-        safe_name = "".join(c for c in resource_name if c.isalnum() or c in (' ', '-', '_')).rstrip()
-        safe_name = safe_name.replace(' ', '_')
-        
-        file_path = self.catalog_dir / "resources" / f"{safe_name}.json"
-        
-        # Structure the analysis data
-        structured_analysis = {
-            "name": resource_name,
-            "url": analysis.get("url", ""),
-            "base_url": analysis.get("base_url", ""),
-            "version": analysis.get("version", ""),
-            "title": analysis.get("title", ""),
-            "description": analysis.get("description", ""),
-            "summary": analysis.get("summary", ""),
-            "endpoints": analysis.get("endpoints", []),
-            "schemas": analysis.get("schemas", {}),
-            "error": analysis.get("error", None),
-            "raw_analysis": analysis  # Keep original analysis for reference
-        }
-        
-        with open(file_path, 'w', encoding='utf-8') as f:
-            json.dump(structured_analysis, f, indent=2, ensure_ascii=False)
-        
-        return str(file_path)
-    
-    def load_tool_schema(self, tool_name: str) -> Optional[Dict[str, Any]]:
-        """Load tool schema from disk."""
-        file_path = self.catalog_dir / "tools" / f"{tool_name}.json"
-        
-        if file_path.exists():
-            with open(file_path, 'r', encoding='utf-8') as f:
-                return json.load(f)
-        return None
-    
-    def load_resource_analysis(self, resource_name: str) -> Optional[Dict[str, Any]]:
-        """Load resource analysis from disk."""
-        safe_name = "".join(c for c in resource_name if c.isalnum() or c in (' ', '-', '_')).rstrip()
-        safe_name = safe_name.replace(' ', '_')
-        
-        file_path = self.catalog_dir / "resources" / f"{safe_name}.json"
-        
-        if file_path.exists():
-            with open(file_path, 'r', encoding='utf-8') as f:
-                return json.load(f)
-        return None
-    
-    def list_cataloged_tools(self) -> List[str]:
-        """List all cataloged tools."""
-        tools_dir = self.catalog_dir / "tools"
-        if not tools_dir.exists():
-            return []
-        
-        tools = []
-        for file_path in tools_dir.glob("*.json"):
-            tools.append(file_path.stem)
-        return tools
-    
-    def list_cataloged_resources(self) -> List[str]:
-        """List all cataloged resources."""
-        resources_dir = self.catalog_dir / "resources"
-        if not resources_dir.exists():
-            return []
-        
-        resources = []
-        for file_path in resources_dir.glob("*.json"):
-            resources.append(file_path.stem)
-        return resources
-    
-    def get_catalog_summary(self) -> Dict[str, Any]:
-        """Get a summary of the catalog contents."""
-        return {
-            "tools": self.list_cataloged_tools(),
-            "resources": self.list_cataloged_resources(),
-            "catalog_dir": str(self.catalog_dir)
-        } 

backup_v1/mcp/catalog_utils.py

@@ -1,147 +0,0 @@
-import json
-from typing import Dict, List, Any, Optional
-from .catalog import MCPCatalog
-
-
-class CatalogUtils:
-    """Utility functions for working with cataloged MCP data."""
-    
-    def __init__(self, catalog_dir: str = "mcp_catalog"):
-        self.catalog = MCPCatalog(catalog_dir)
-    
-    def get_tool_schema(self, tool_name: str) -> Optional[Dict[str, Any]]:
-        """Get the schema for a specific tool."""
-        return self.catalog.load_tool_schema(tool_name)
-    
-    def get_resource_analysis(self, resource_name: str) -> Optional[Dict[str, Any]]:
-        """Get the analysis for a specific resource."""
-        return self.catalog.load_resource_analysis(resource_name)
-    
-    def list_available_tools(self) -> List[str]:
-        """List all available tools in the catalog."""
-        return self.catalog.list_cataloged_tools()
-    
-    def list_available_resources(self) -> List[str]:
-        """List all available resources in the catalog."""
-        return self.catalog.list_cataloged_resources()
-    
-    def get_tool_input_schema(self, tool_name: str) -> Optional[Dict[str, Any]]:
-        """Get the input schema for a tool."""
-        schema = self.get_tool_schema(tool_name)
-        if schema:
-            return schema.get("inputSchema", {})
-        return None
-    
-    def get_tool_output_schema(self, tool_name: str) -> Optional[Dict[str, Any]]:
-        """Get the output schema for a tool."""
-        schema = self.get_tool_schema(tool_name)
-        if schema:
-            return schema.get("outputSchema", {})
-        return None
-    
-    def get_tool_description(self, tool_name: str) -> Optional[str]:
-        """Get the description for a tool."""
-        schema = self.get_tool_schema(tool_name)
-        if schema:
-            return schema.get("description", "")
-        return None
-    
-    def format_tool_request(self, tool_name: str, params: Dict[str, Any]) -> Dict[str, Any]:
-        """Format a request for a tool based on its schema."""
-        schema = self.get_tool_schema(tool_name)
-        if not schema:
-            raise ValueError(f"Tool '{tool_name}' not found in catalog")
-        
-        # Validate required parameters
-        input_schema = schema.get("inputSchema", {})
-        required_props = input_schema.get("required", [])
-        
-        missing_params = []
-        for prop in required_props:
-            if prop not in params:
-                missing_params.append(prop)
-        
-        if missing_params:
-            raise ValueError(f"Missing required parameters for {tool_name}: {missing_params}")
-        
-        return {
-            "method": "tools/call",
-            "params": {
-                "name": tool_name,
-                "arguments": params
-            }
-        }
-    
-    def get_api_endpoints(self, resource_name: str) -> List[Dict[str, Any]]:
-        """Get API endpoints from a resource analysis."""
-        analysis = self.get_resource_analysis(resource_name)
-        if analysis:
-            return analysis.get("endpoints", [])
-        return []
-    
-    def get_api_schemas(self, resource_name: str) -> Dict[str, Any]:
-        """Get API schemas from a resource analysis."""
-        analysis = self.get_resource_analysis(resource_name)
-        if analysis:
-            return analysis.get("schemas", {})
-        return {}
-    
-    def find_api_endpoint(self, resource_name: str, path: str, method: str = "GET") -> Optional[Dict[str, Any]]:
-        """Find a specific API endpoint in a resource."""
-        endpoints = self.get_api_endpoints(resource_name)
-        for endpoint in endpoints:
-            if endpoint.get("path") == path and endpoint.get("method") == method.upper():
-                return endpoint
-        return None
-    
-    def get_catalog_summary(self) -> Dict[str, Any]:
-        """Get a summary of the catalog contents."""
-        return self.catalog.get_catalog_summary()
-    
-    def print_tool_info(self, tool_name: str):
-        """Print detailed information about a tool."""
-        schema = self.get_tool_schema(tool_name)
-        if not schema:
-            print(f"❌ Tool '{tool_name}' not found in catalog")
-            return
-        
-        print(f"\n🔧 Tool: {tool_name}")
-        print(f"📝 Description: {schema.get('description', 'No description')}")
-        
-        input_schema = schema.get("inputSchema", {})
-        if input_schema:
-            print(f"📥 Input Schema:")
-            properties = input_schema.get("properties", {})
-            required = input_schema.get("required", [])
-            
-            for prop_name, prop_info in properties.items():
-                required_mark = " (required)" if prop_name in required else ""
-                print(f"  - {prop_name}: {prop_info.get('type', 'unknown')}{required_mark}")
-                if prop_info.get("description"):
-                    print(f"    Description: {prop_info['description']}")
-    
-    def print_resource_info(self, resource_name: str):
-        """Print detailed information about a resource."""
-        analysis = self.get_resource_analysis(resource_name)
-        if not analysis:
-            print(f"❌ Resource '{resource_name}' not found in catalog")
-            return
-        
-        print(f"\n📚 Resource: {resource_name}")
-        print(f"📝 Title: {analysis.get('title', 'No title')}")
-        print(f"🔗 URL: {analysis.get('url', 'No URL')}")
-        print(f"📋 Summary: {analysis.get('summary', 'No summary')}")
-        
-        endpoints = analysis.get("endpoints", [])
-        if endpoints:
-            print(f"🔌 Endpoints ({len(endpoints)}):")
-            for endpoint in endpoints[:5]:  # Show first 5
-                print(f"  - {endpoint.get('method', 'UNKNOWN')} {endpoint.get('path', 'unknown')}")
-                if endpoint.get("summary"):
-                    print(f"    Summary: {endpoint['summary']}")
-        
-        schemas = analysis.get("schemas", {})
-        if schemas:
-            print(f"📋 Schemas ({len(schemas)}):")
-            for schema_name in list(schemas.keys())[:5]:  # Show first 5
-                print(f"  - {schema_name}") 

backup_v1/mcp/compact_utils.py

@@ -1,216 +0,0 @@
-import json
-import asyncio
-import aiohttp
-import re
-from typing import Dict, List, Any, Optional
-from pathlib import Path
-from app.config.api_endpoints import get_base_url
-from .catalog_utils import CatalogUtils
-
-
-class CompactSpecsUtils:
-    """Utility functions for working with compact API specs."""
-    
-    def __init__(self, catalog_dir: str = "mcp_catalog"):
-        self.catalog_utils = CatalogUtils(catalog_dir)
-        # Import the generated compact specs
-        try:
-            from app.config.compact_api_specs import COMPACT_API_SPECS
-            self.compact_specs = COMPACT_API_SPECS
-        except ImportError:
-            self.compact_specs = {}
-            print("⚠️ Compact API specs not found. Run initialization first.")
-    
-    def get_working_tools(self) -> List[str]:
-        """Get list of working tools from compact specs."""
-        return [name for name, spec in self.compact_specs.items() 
-                if "parameters" in spec]  # Tools have parameters
-    
-    def get_working_resources(self) -> List[str]:
-        """Get list of working resources from compact specs."""
-        return [name for name, spec in self.compact_specs.items() 
-                if "total_endpoints" in spec]  # Resources have endpoints
-    
-    def get_tool_spec(self, tool_name: str) -> Optional[Dict[str, Any]]:
-        """Get compact spec for a tool."""
-        if tool_name in self.compact_specs and "parameters" in self.compact_specs[tool_name]:
-            return self.compact_specs[tool_name]
-        return None
-    
-    def get_resource_spec(self, resource_name: str) -> Optional[Dict[str, Any]]:
-        """Get compact spec for a resource."""
-        if resource_name in self.compact_specs and "total_endpoints" in self.compact_specs[resource_name]:
-            return self.compact_specs[resource_name]
-        return None
-    
-    def format_tool_request(self, tool_name: str, params: Dict[str, Any]) -> Dict[str, Any]:
-        """Format a tool request using compact specs."""
-        spec = self.get_tool_spec(tool_name)
-        if not spec:
-            raise ValueError(f"Tool '{tool_name}' not found in compact specs")
-        
-        # Validate parameters against compact spec
-        available_params = spec.get("parameters", {})
-        unknown_params = [p for p in params.keys() if p not in available_params]
-        
-        if unknown_params:
-            print(f"⚠️ Warning: Unknown parameters for {tool_name}: {unknown_params}")
-        
-        return {
-            "method": "tools/call",
-            "params": {
-                "name": tool_name,
-                "arguments": params
-            }
-        }
-    
-    def format_resource_request(self, resource_name: str, endpoint_path: str, method: str = "GET", params: Dict[str, Any] = None) -> Dict[str, Any]:
-        """Format a resource (API) request using compact specs."""
-        spec = self.get_resource_spec(resource_name)
-        if not spec:
-            raise ValueError(f"Resource '{resource_name}' not found in compact specs")
-        
-        # Validate endpoint exists
-        key_endpoints = spec.get("key_endpoints", [])
-        endpoint_found = any(
-            ep.get("path") == endpoint_path and ep.get("method") == method 
-            for ep in key_endpoints
-        )
-        
-        if not endpoint_found:
-            print(f"⚠️ Warning: Endpoint {method} {endpoint_path} not found in key endpoints for {resource_name}")
-        
-        return {
-            "method": "resources/read",
-            "params": {
-                "name": resource_name,
-                "uri": endpoint_path,
-                "arguments": params or {}
-            }
-        }
-    
-    async def make_direct_api_call(self, resource_name: str, endpoint_path: str, method: str = "GET", params: Dict[str, Any] = None) -> Dict[str, Any]:
-        """Make a direct API call using hardcoded base URLs."""
-        # Use hardcoded base URL
-        base_url = get_base_url(resource_name)
-        if not base_url:
-            raise ValueError(f"No base URL found for resource '{resource_name}'")
-        
-        # Construct the full URL
-        full_url = f"{base_url.rstrip('/')}{endpoint_path}"
-        
-        # Make the API call
-        async with aiohttp.ClientSession() as session:
-            if method.upper() == "GET":
-                # For GET requests, add params to query string
-                async with session.get(full_url, params=params or {}) as resp:
-                    return {
-                        "status": resp.status,
-                        "data": await resp.json() if resp.status == 200 else None,
-                        "error": await resp.text() if resp.status != 200 else None
-                    }
-            else:
-                # For other methods, send params in body
-                async with session.request(method, full_url, json=params or {}) as resp:
-                    return {
-                        "status": resp.status,
-                        "data": await resp.json() if resp.status == 200 else None,
-                        "error": await resp.text() if resp.status != 200 else None
-                    }
-    
-    def get_resource_endpoint_info(self, resource_name: str, endpoint_path: str, method: str = "GET") -> Optional[Dict[str, Any]]:
-        """Get information about a specific endpoint in a resource."""
-        spec = self.get_resource_spec(resource_name)
-        if not spec:
-            return None
-        
-        key_endpoints = spec.get("key_endpoints", [])
-        for endpoint in key_endpoints:
-            if endpoint.get("path") == endpoint_path and endpoint.get("method") == method:
-                return endpoint
-        return None
-    
-    def get_tool_parameters(self, tool_name: str) -> Dict[str, str]:
-        """Get available parameters for a tool."""
-        spec = self.get_tool_spec(tool_name)
-        if spec:
-            return spec.get("parameters", {})
-        return {}
-    
-    def get_resource_endpoints(self, resource_name: str) -> List[Dict[str, str]]:
-        """Get key endpoints for a resource."""
-        spec = self.get_resource_spec(resource_name)
-        if spec:
-            return spec.get("key_endpoints", [])
-        return []
-    
-    def print_compact_summary(self):
-        """Print a summary of compact specs."""
-        tools = self.get_working_tools()
-        resources = self.get_working_resources()
-        
-        print(f"📊 Compact Specs Summary:")
-        print(f"  🔧 Working Tools: {len(tools)}")
-        for tool_name in tools:
-            spec = self.get_tool_spec(tool_name)
-            param_count = len(spec.get("parameters", {}))
-            print(f"    - {tool_name} ({param_count} parameters)")
-        
-        print(f"  📚 Working Resources: {len(resources)}")
-        for resource_name in resources:
-            spec = self.get_resource_spec(resource_name)
-            total_endpoints = spec.get("total_endpoints", 0)
-            print(f"    - {resource_name} ({total_endpoints} endpoints)")
-    
-    def print_tool_info(self, tool_name: str):
-        """Print detailed information about a tool from compact specs."""
-        spec = self.get_tool_spec(tool_name)
-        if not spec:
-            print(f"❌ Tool '{tool_name}' not found in compact specs")
-            return
-        
-        print(f"\n🔧 Tool: {tool_name}")
-        print(f"📝 Description: {spec.get('description', 'No description')}")
-        
-        parameters = spec.get("parameters", {})
-        if parameters:
-            print(f"📥 Parameters:")
-            for param_name, param_desc in parameters.items():
-                print(f"  - {param_name}: {param_desc}")
-    
-    def print_resource_info(self, resource_name: str):
-        """Print detailed information about a resource from compact specs."""
-        spec = self.get_resource_spec(resource_name)
-        if not spec:
-            print(f"❌ Resource '{resource_name}' not found in compact specs")
-            return
-        
-        print(f"\n📚 Resource: {resource_name}")
-        print(f"📝 Description: {spec.get('description', 'No description')}")
-        print(f"🔌 Total Endpoints: {spec.get('total_endpoints', 0)}")
-        
-        key_endpoints = spec.get("key_endpoints", [])
-        if key_endpoints:
-            print(f"🔌 Key Endpoints:")
-            for endpoint in key_endpoints:
-                print(f"  - {endpoint.get('method', 'UNKNOWN')} {endpoint.get('path', 'unknown')}")
-    
-    def find_tool_by_description(self, keyword: str) -> List[str]:
-        """Find tools by keyword in description."""
-        matching_tools = []
-        for tool_name, spec in self.compact_specs.items():
-            if "parameters" in spec:  # It's a tool
-                description = spec.get("description", "").lower()
-                if keyword.lower() in description:
-                    matching_tools.append(tool_name)
-        return matching_tools
-    
-    def find_resource_by_description(self, keyword: str) -> List[str]:
-        """Find resources by keyword in description."""
-        matching_resources = []
-        for resource_name, spec in self.compact_specs.items():
-            if "total_endpoints" in spec:  # It's a resource
-                description = spec.get("description", "").lower()
-                if keyword.lower() in description:
-                    matching_resources.append(resource_name)
-        return matching_resources 

backup_v1/mcp/specs_generator.py

@@ -1,487 +0,0 @@
-import json
-import asyncio
-from typing import Dict, List, Any, Optional
-from .catalog_utils import CatalogUtils
-from .stream_client import load_server_config, StreamableHttpMCPClient
-from app.config.settings import EXCLUDE_AUTH_REQUIRED_ENDPOINTS, EXCLUDE_401_RESPONSES, STRICT_AUTH_FILTERING
-
-
-class SpecsGenerator:
-    """Generates compact API specs from cataloged data and tests functionality."""
-    
-    def __init__(self, catalog_dir: str = "mcp_catalog"):
-        self.catalog_utils = CatalogUtils(catalog_dir)
-        self.compact_specs = {}
-    
-    async def test_tool_functionality(self, tool_name: str) -> bool:
-        """Test if a tool is working by making a minimal call."""
-        try:
-            server_config = load_server_config()
-            async with StreamableHttpMCPClient(server_config) as client:
-                await client.initialize()
-                
-                # Get tool schema to find a valid test parameter
-                schema = self.catalog_utils.get_tool_schema(tool_name)
-                if not schema:
-                    return False
-                
-                input_schema = schema.get("inputSchema", {})
-                properties = input_schema.get("properties", {})
-                
-                # Find a simple parameter to test with
-                test_params = {}
-                for param_name, param_info in properties.items():
-                    if param_info.get("type") == "string" and not param_info.get("required", False):
-                        test_params[param_name] = "test"
-                        break
-                    elif param_info.get("type") == "number" and not param_info.get("required", False):
-                        test_params[param_name] = 1
-                        break
-                
-                # Make a test call
-                request = self.catalog_utils.format_tool_request(tool_name, test_params)
-                result = await client.call("tools/call", request["params"])
-                
-                # Check if we got a valid response (not an error)
-                return "error" not in result
-                
-        except Exception as e:
-            print(f"❌ Tool {tool_name} test failed: {e}")
-            return False
-    
-    async def test_resource_functionality(self, resource_name: str) -> bool:
-        """Test if a resource (API) is working by checking its endpoints."""
-        try:
-            analysis = self.catalog_utils.get_resource_analysis(resource_name)
-            if not analysis:
-                return False
-            
-            # Check if the analysis has errors
-            if analysis.get("error"):
-                return False
-            
-            # Check if we have endpoints
-            endpoints = analysis.get("endpoints", [])
-            if not endpoints:
-                return False
-            
-            # Check if the API has meaningful content
-            title = analysis.get("title", "")
-            description = analysis.get("description", "")
-            
-            # If it's a basic resource without API content, consider it working
-            if not title and not description and len(endpoints) == 0:
-                return True
-            
-            # For API resources, check if we have valid endpoints
-            return len(endpoints) > 0
-            
-        except Exception as e:
-            print(f"❌ Resource {resource_name} test failed: {e}")
-            return False
-    
-    def generate_compact_tool_spec(self, tool_name: str) -> Optional[Dict[str, Any]]:
-        """Generate compact spec for a tool."""
-        schema = self.catalog_utils.get_tool_schema(tool_name)
-        if not schema:
-            return None
-        
-        description = schema.get("description", "")
-        input_schema = schema.get("inputSchema", {})
-        properties = input_schema.get("properties", {})
-        required = input_schema.get("required", [])
-        
-        # Extract ALL parameters with comprehensive details
-        parameters = {}
-        for param_name, param_info in properties.items():
-            param_type = param_info.get("type", "string")
-            param_desc = param_info.get("description", "")
-            param_enum = param_info.get("enum", [])
-            param_required = param_name in required
-            param_items = param_info.get("items", {})
-            param_format = param_info.get("format", "")
-            param_default = param_info.get("default")
-            
-            # Build comprehensive parameter description
-            desc_parts = []
-            
-            # Add base description
-            if param_desc:
-                base_desc = param_desc.split('.')[0] if '.' in param_desc else param_desc
-                desc_parts.append(base_desc)
-            
-            # Add type information
-            if param_type == "array" and param_items:
-                item_type = param_items.get("type", "string")
-                desc_parts.append(f"Array of {item_type}")
-            else:
-                desc_parts.append(f"Type: {param_type}")
-            
-            # Add format if specified
-            if param_format:
-                desc_parts.append(f"Format: {param_format}")
-            
-            # Add required/optional status
-            desc_parts.append("Required" if param_required else "Optional")
-            
-            # Add enum values if available
-            if param_enum:
-                enum_str = ", ".join([f'"{val}"' for val in param_enum[:8]])  # Show up to 8 values
-                if len(param_enum) > 8:
-                    enum_str += f"... (and {len(param_enum) - 8} more)"
-                desc_parts.append(f"Valid values: {enum_str}")
-            
-            # Add default value if specified
-            if param_default is not None:
-                desc_parts.append(f"Default: {param_default}")
-            
-            # Combine all parts
-            desc = ". ".join(desc_parts)
-            
-            # Clean up description for Python string - escape all special characters
-            desc = (desc.replace('\\', '\\\\')  # Escape backslashes first
-                       .replace('"', '\\"')    # Escape quotes
-                       .replace('\n', ' ')     # Replace newlines
-                       .replace('\r', ' ')     # Replace carriage returns
-                       .replace('`', "'"))     # Replace backticks with single quotes
-            parameters[param_name] = desc
-        
-        # Clean up description for Python string
-        clean_description = description[:200] + "..." if len(description) > 200 else description
-        clean_description = clean_description.replace('"', '\\"').replace('\n', ' ').replace('\r', ' ')
-        
-        return {
-            "description": clean_description,
-            "parameters": parameters
-        }
-    
-    def _extract_endpoint_parameters(self, endpoint: Optional[Dict[str, Any]]) -> Dict[str, str]:
-        """Extract comprehensive parameter information from an OpenAPI endpoint."""
-        if not endpoint:
-            return {}
-        
-        parameters_info = {}
-        endpoint_params = endpoint.get("parameters", [])
-        
-        for param in endpoint_params:
-            param_name = param.get("name", "")
-            param_in = param.get("in", "")
-            param_type = param.get("type", "string")
-            param_desc = param.get("description", "")
-            param_required = param.get("required", False)
-            param_enum = param.get("enum", [])
-            param_format = param.get("format", "")
-            param_default = param.get("default")
-            
-            # Skip path parameters for base URL approach
-            if param_in == "path":
-                continue
-            
-            # Skip body parameters as they're not query params
-            if param_in == "body":
-                continue
-            
-            # Build comprehensive parameter description
-            desc_parts = []
-            
-            # Add base description
-            if param_desc:
-                # Clean up description - remove excessive details but keep key info
-                clean_desc = param_desc.replace('\n', ' ').replace('\r', ' ')
-                if len(clean_desc) > 200:
-                    # Take first 200 chars but try to end at sentence
-                    truncated = clean_desc[:200]
-                    last_period = truncated.rfind('.')
-                    if last_period > 100:  # If there's a period in reasonable range
-                        clean_desc = truncated[:last_period + 1]
-                    else:
-                        clean_desc = truncated + "..."
-                desc_parts.append(clean_desc)
-            
-            # Add type and location information
-            desc_parts.append(f"Type: {param_type}")
-            if param_in:
-                desc_parts.append(f"Location: {param_in}")
-            
-            # Add required/optional status
-            desc_parts.append("Required" if param_required else "Optional")
-            
-            # Add format if specified
-            if param_format:
-                desc_parts.append(f"Format: {param_format}")
-            
-            # Add enum values if available
-            if param_enum:
-                enum_str = ", ".join([f'"{val}"' for val in param_enum[:6]])  # Show up to 6 values
-                if len(param_enum) > 6:
-                    enum_str += f"... (and {len(param_enum) - 6} more)"
-                desc_parts.append(f"Valid values: {enum_str}")
-            
-            # Add default value if specified
-            if param_default is not None:
-                desc_parts.append(f"Default: {param_default}")
-            
-            # Combine all parts
-            desc = ". ".join(desc_parts)
-            
-            # Clean up description for Python string - escape all special characters
-            desc = (desc.replace('\\', '\\\\')  # Escape backslashes first
-                       .replace('"', '\\"')    # Escape quotes
-                       .replace('\n', ' ')     # Replace newlines
-                       .replace('\r', ' ')     # Replace carriage returns
-                       .replace('`', "'"))     # Replace backticks with single quotes
-            
-            if param_name:
-                parameters_info[param_name] = desc
-        
-        return parameters_info
-    
-    def generate_compact_resource_spec(self, resource_name: str) -> Optional[Dict[str, Any]]:
-        """Generate compact spec for a resource."""
-        analysis = self.catalog_utils.get_resource_analysis(resource_name)
-        if not analysis:
-            return None
-        
-        title = analysis.get("title", "")
-        description = analysis.get("description", "")
-        endpoints = analysis.get("endpoints", [])
-        base_url = analysis.get("base_url", "")
-        
-        # Filter out authentication-required endpoints if configured
-        if EXCLUDE_AUTH_REQUIRED_ENDPOINTS:
-            original_count = len(endpoints)
-            filtered_endpoints = []
-            
-            for e in endpoints:
-                requires_auth = e.get("requires_auth", False)
-                
-                # Apply different filtering strategies based on configuration
-                should_exclude = False
-                
-                if EXCLUDE_401_RESPONSES and requires_auth:
-                    should_exclude = True
-                elif STRICT_AUTH_FILTERING:
-                    # More strict filtering - exclude if any auth indicators found
-                    description = e.get("description", "").lower()
-                    summary = e.get("summary", "").lower()
-                    auth_keywords = ["authentication", "authorization", "bearer", "token", "auth", "login", "credential", "secured", "private", "admin"]
-                    should_exclude = any(keyword in description or keyword in summary for keyword in auth_keywords)
-                
-                if not should_exclude:
-                    filtered_endpoints.append(e)
-            
-            filtered_count = len(filtered_endpoints)
-            excluded_count = original_count - filtered_count
-            print(f"    📊 Filtered {excluded_count} auth-required endpoints from {resource_name} ({original_count} → {filtered_count})")
-        else:
-            filtered_endpoints = endpoints
-
-        # Extract key endpoints based on user requirements (base URL approach)
-        key_endpoints = []
-        
-        # For Member API - use base /members endpoint with comprehensive parameters
-        if "member" in resource_name.lower():
-            # Find the main /members endpoint - check both filtered and unfiltered endpoints
-            # since /members is known to work without auth despite having 401 responses
-            members_endpoint = None
-            
-            # First check filtered endpoints
-            for endpoint in filtered_endpoints:
-                if endpoint.get("path") == "/members" and endpoint.get("method") == "GET":
-                    members_endpoint = endpoint
-                    break
-            
-            # If not found in filtered, check original endpoints (known working case)
-            if not members_endpoint:
-                for endpoint in endpoints:
-                    if endpoint.get("path") == "/members" and endpoint.get("method") == "GET":
-                        # Special case: /members works without auth despite 401 responses
-                        if "authentication credential is optional" in endpoint.get("description", "").lower():
-                            members_endpoint = endpoint
-                            break
-            
-            # Extract comprehensive parameter information
-            params_info = self._extract_endpoint_parameters(members_endpoint) if members_endpoint else {}
-            params_desc = f"Available parameters: {', '.join(params_info.keys())}" if params_info else "Get members list with optional filters like handle, page, perPage"
-            
-            key_endpoints.append({
-                "path": "/members",
-                "method": "GET",
-                "full_url": f"{base_url}/members",
-                "description": params_desc,
-                "parameters": params_info
-            })
-        
-        # For Challenges API - use base /challenges endpoint with comprehensive parameters
-        elif "challenge" in resource_name.lower():
-            # Find the main /challenges endpoint to extract real parameters
-            challenges_endpoint = None
-            for endpoint in filtered_endpoints:
-                if endpoint.get("path") == "/challenges" and endpoint.get("method") == "GET":
-                    challenges_endpoint = endpoint
-                    break
-            
-            # Extract comprehensive parameter information
-            params_info = self._extract_endpoint_parameters(challenges_endpoint) if challenges_endpoint else {}
-            params_desc = f"Available parameters: {', '.join(params_info.keys())}" if params_info else "Get challenges list with optional filters like status, track, page, perPage"
-            
-            key_endpoints.append({
-                "path": "/challenges",
-                "method": "GET", 
-                "full_url": f"{base_url}/challenges",
-                "description": params_desc,
-                "parameters": params_info
-            })
-            
-        # For other APIs - extract important base endpoints
-        else:
-            seen_base_paths = set()
-            for endpoint in filtered_endpoints[:5]:  # Limit to first 5
-                path = endpoint.get("path", "")
-                method = endpoint.get("method", "GET")
-                summary = endpoint.get("summary", "")
-                full_url = endpoint.get("full_url", "")
-                
-                # Extract base path (remove path parameters)
-                base_path = path.split('/')[1] if '/' in path and len(path.split('/')) > 1 else path
-                base_path = f"/{base_path}" if not base_path.startswith('/') else base_path
-                
-                if base_path not in seen_base_paths and path and method:
-                    seen_base_paths.add(base_path)
-                    key_endpoints.append({
-                        "path": path,
-                        "method": method,
-                        "summary": summary[:100] + "..." if len(summary) > 100 else summary,
-                        "full_url": full_url
-                    })
-        
-        # Generate compact description
-        if title and description:
-            compact_desc = f"{title}: {description[:150]}..."
-        elif title:
-            compact_desc = title
-        elif description:
-            compact_desc = description[:200] + "..."
-        else:
-            compact_desc = f"API with {len(filtered_endpoints)} endpoints"
-        
-        # Clean up description for Python string
-        compact_desc = compact_desc.replace('"', '\\"').replace('\n', ' ').replace('\r', ' ')
-        
-        return {
-            "description": compact_desc,
-            "key_endpoints": key_endpoints,
-            "total_endpoints": len(filtered_endpoints),
-            "base_url": base_url
-        }
-    
-    async def generate_compact_specs(self) -> Dict[str, Any]:
-        """Generate compact specs for all working tools and resources."""
-        print("🔍 Testing functionality and generating compact specs...")
-        
-        # Test and generate tool specs
-        tools = self.catalog_utils.list_available_tools()
-        working_tools = {}
-        
-        for tool_name in tools:
-            print(f"  🧪 Testing tool: {tool_name}")
-            if await self.test_tool_functionality(tool_name):
-                spec = self.generate_compact_tool_spec(tool_name)
-                if spec:
-                    working_tools[tool_name] = spec
-                    print(f"    ✅ {tool_name} - Working")
-            else:
-                print(f"    ❌ {tool_name} - Not working")
-        
-        # Test and generate resource specs
-        resources = self.catalog_utils.list_available_resources()
-        working_resources = {}
-        
-        for resource_name in resources:
-            print(f"  🧪 Testing resource: {resource_name}")
-            if await self.test_resource_functionality(resource_name):
-                spec = self.generate_compact_resource_spec(resource_name)
-                if spec:
-                    working_resources[resource_name] = spec
-                    print(f"    ✅ {resource_name} - Working")
-            else:
-                print(f"    ❌ {resource_name} - Not working")
-        
-        return {
-            "tools": working_tools,
-            "resources": working_resources
-        }
-    
-    def save_compact_specs(self, specs: Dict[str, Any], output_file: str = "app/config/compact_api_specs.py"):
-        """Save compact specs to a Python file."""
-        with open(output_file, 'w', encoding='utf-8') as f:
-            f.write("# Generated compact API specs from MCP catalog\n")
-            f.write("# Only includes working tools and resources\n\n")
-            
-            f.write("COMPACT_API_SPECS = {\n")
-            
-            # Write tools
-            f.write("    # Working Tools\n")
-            for tool_name, spec in specs.get("tools", {}).items():
-                f.write(f'    "{tool_name}": {{\n')
-                f.write(f'        "description": "{spec["description"]}",\n')
-                f.write("        \"parameters\": {\n")
-                for param_name, param_desc in spec["parameters"].items():
-                    f.write(f'            "{param_name}": "{param_desc}",\n')
-                f.write("        }\n")
-                f.write("    },\n")
-            
-            # Write resources
-            f.write("\n    # Working Resources\n")
-            for resource_name, spec in specs.get("resources", {}).items():
-                f.write(f'    "{resource_name}": {{\n')
-                f.write(f'        "description": "{spec["description"]}",\n')
-                f.write(f'        "total_endpoints": {spec["total_endpoints"]},\n')
-                if spec.get("base_url"):
-                    f.write(f'        "base_url": "{spec["base_url"]}",\n')
-                f.write("        \"key_endpoints\": [\n")
-                for endpoint in spec["key_endpoints"][:5]:  # Limit to 5 key endpoints
-                    endpoint_line = f'            {{"path": "{endpoint["path"]}", "method": "{endpoint["method"]}"'
-                    if endpoint.get("full_url"):
-                        endpoint_line += f', "full_url": "{endpoint["full_url"]}"'
-                    if endpoint.get("description"):
-                        # Escape description for Python string
-                        escaped_desc = endpoint["description"].replace('"', '\\"')
-                        endpoint_line += f', "description": "{escaped_desc}"'
-                    if endpoint.get("parameters"):
-                        # Write parameters as a nested dictionary
-                        endpoint_line += ', "parameters": {'
-                        param_entries = []
-                        for param_name, param_desc in endpoint["parameters"].items():
-                            # Escape all special characters for Python string
-                            escaped_param_desc = (param_desc.replace('\\', '\\\\')  # Escape backslashes first
-                                                             .replace('"', '\\"')    # Escape quotes
-                                                             .replace('\n', ' ')     # Replace newlines
-                                                             .replace('\r', ' ')     # Replace carriage returns
-                                                             .replace('`', "'"))     # Replace backticks with single quotes
-                            param_entries.append(f'"{param_name}": "{escaped_param_desc}"')
-                        endpoint_line += ', '.join(param_entries)
-                        endpoint_line += '}'
-                    endpoint_line += "},\n"
-                    f.write(endpoint_line)
-                f.write("        ]\n")
-                f.write("    },\n")
-            
-            f.write("}\n")
-        
-        print(f"✅ Compact specs saved to {output_file}")
-    
-    def print_summary(self, specs: Dict[str, Any]):
-        """Print a summary of the generated specs."""
-        tools = specs.get("tools", {})
-        resources = specs.get("resources", {})
-        
-        print(f"\n📊 Compact Specs Summary:")
-        print(f"  🔧 Working Tools: {len(tools)}")
-        for tool_name in tools:
-            print(f"    - {tool_name}")
-        
-        print(f"  📚 Working Resources: {len(resources)}")
-        for resource_name in resources:
-            total_endpoints = resources[resource_name].get("total_endpoints", 0)
-            print(f"    - {resource_name} ({total_endpoints} endpoints)") 

backup_v1/mcp/stream_client.py

@@ -1,308 +0,0 @@
-import os
-import asyncio
-import json
-import uuid
-from dataclasses import dataclass
-from typing import Any, Dict, List, Optional, Tuple
-import aiohttp
-from app.config import settings
-
-# Default to local config file instead of ~/.cursor/mcp.json
-DEFAULT_CONFIG_PATH = "mcp-config.json"
-DEFAULT_SERVER_NAME = "topcoder"
-DEFAULT_PROTOCOL_VERSION = "2024-11-05"
-
-
-@dataclass
-class ServerConfig:
-    name: str
-    url: str
-    headers: Dict[str, str]
-
-
-def load_server_config(
-    server_name: str = DEFAULT_SERVER_NAME,
-    config_path: str = DEFAULT_CONFIG_PATH,
-) -> ServerConfig:
-    # Allow override via env var for quick testing
-    env_url = os.environ.get("MCP_SERVER_URL")
-    env_headers: Dict[str, str] = {}
-    if token := os.environ.get("MCP_TOKEN"):
-        env_headers["Authorization"] = f"Bearer {token}"
-
-    # Try config file
-    url_from_config: Optional[str] = None
-    headers_from_config: Dict[str, str] = {}
-
-    if os.path.isfile(config_path):
-        try:
-            with open(config_path, "r", encoding="utf-8") as f:
-                cfg = json.load(f)
-            servers = cfg.get("mcpServers", {})
-            server_entry = servers.get(server_name, {})
-            url_from_config = server_entry.get("url")
-            # Common optional fields others may use
-            api_key = server_entry.get("apiKey")
-            if api_key:
-                headers_from_config["Authorization"] = f"Bearer {api_key}"
-            # Raw headers block if present
-            for header_key, header_value in server_entry.get("headers", {}).items():
-                headers_from_config[header_key] = header_value
-        except Exception as exc:
-            raise RuntimeError(f"Failed to read config {config_path}: {exc}")
-    else:
-        print(f"Warning: Config file {config_path} not found. Using environment variables or defaults.")
-
-    # Final resolution order: env > config > default
-    final_url = env_url or url_from_config or "https://api.topcoder-dev.com/v6/mcp/mcp"
-    final_headers = {**headers_from_config, **env_headers}
-
-    return ServerConfig(name=server_name, url=final_url, headers=final_headers)
-
-
-def build_json_rpc_request(method: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
-    request_id = str(uuid.uuid4())
-    return {
-        "jsonrpc": "2.0",
-        "id": request_id,
-        "method": method,
-        "params": params or {},
-    }
-
-
-def _parse_streamable_http_body(text_body: str, content_type: str) -> Dict[str, Any]:
-    """Parse a Streamable HTTP response body that may be JSON or SSE-framed.
-
-    Returns a single JSON-RPC message object (not an array). If the server
-    wraps responses in an array, the first element will be returned.
-    """
-    ct_lower = (content_type or "").lower()
-
-    # Try JSON first when content-type hints JSON
-    if "application/json" in ct_lower:
-        data = json.loads(text_body)
-        if isinstance(data, list) and data:
-            return data[0]
-        return data
-
-    # Fallback: detect SSE-style content or bodies that include 'event:' lines
-    if "text/event-stream" in ct_lower or text_body.lstrip().startswith("event:"):
-        # Collect the first 'data: ' block
-        json_payload_lines: list[str] = []
-        for raw_line in text_body.splitlines():
-            line = raw_line.strip("\r\n")
-            if line.startswith("data:"):
-                # After 'data:' there may be a space; strip and collect
-                # Everything after 'data:' is part of the JSON payload line
-                json_payload_lines.append(line[len("data:"):].lstrip())
-            elif line == "":
-                # End of event
-                if json_payload_lines:
-                    break
-        if not json_payload_lines:
-            raise json.JSONDecodeError("No data payload found in SSE body", text_body, 0)
-        joined = "\n".join(json_payload_lines)
-        data = json.loads(joined)
-        if isinstance(data, list) and data:
-            return data[0]
-        return data
-
-    # Unknown content-type; try best-effort JSON
-    data = json.loads(text_body)
-    if isinstance(data, list) and data:
-        return data[0]
-    return data
-
-
-class StreamableHttpMCPClient:
-    def __init__(self, server_config: ServerConfig, timeout_seconds: int = 60) -> None:
-        self.server_config = server_config
-        self.timeout_seconds = timeout_seconds
-        self.session: Optional[aiohttp.ClientSession] = None
-        self.mcp_session_id: Optional[str] = None
-
-    async def __aenter__(self) -> "StreamableHttpMCPClient":
-        headers = {"Content-Type": "application/json"}
-        # Per spec, client should accept both json and event-stream
-        headers["Accept"] = "application/json, text/event-stream"
-        # Add configured headers (e.g., Authorization)
-        headers.update(self.server_config.headers)
-        self.session = aiohttp.ClientSession(headers=headers)
-        return self
-
-    async def __aexit__(self, exc_type, exc, tb) -> None:
-        if self.session is not None:
-            await self.session.close()
-            self.session = None
-
-    async def initialize(self) -> Dict[str, Any]:
-        if self.session is None:
-            raise RuntimeError("Client session not started. Use 'async with'.")
-
-        payload = build_json_rpc_request(
-            "initialize",
-            params={
-                "protocolVersion": DEFAULT_PROTOCOL_VERSION,
-                "clientInfo": {"name": "python-mcp-client", "version": "0.1.0"},
-                # It's helpful to advertise capabilities, though not required for list calls
-                "capabilities": {"resources": {}, "prompts": {}, "tools": {}},
-            },
-        )
-        async with self.session.post(self.server_config.url, json=payload, timeout=self.timeout_seconds) as resp:
-            text = await resp.text()
-            try:
-                data = _parse_streamable_http_body(text, resp.headers.get("Content-Type", ""))
-            except json.JSONDecodeError as exc:
-                raise RuntimeError(f"Failed to parse initialize response: {exc}\nRaw: {text}") from exc
-
-            # Capture session ID from headers (case-insensitive)
-            session_id = None
-            for key, value in resp.headers.items():
-                if key.lower() == "mcp-session-id":
-                    session_id = value
-                    break
-            if not session_id:
-                # Fallback to common casings
-                session_id = resp.headers.get("Mcp-Session-Id") or resp.headers.get("MCP-Session-Id")
-
-            if not session_id:
-                raise RuntimeError(
-                    "Missing MCP-Session-Id in initialize response headers. "
-                    f"HTTP {resp.status}, body: {text}"
-                )
-
-            self.mcp_session_id = session_id
-            return data
-
-    async def send_initialized_notification(self) -> Tuple[int, str]:
-        if self.session is None:
-            raise RuntimeError("Client session not started. Use 'async with'.")
-        if not self.mcp_session_id:
-            raise RuntimeError("No MCP session established. Call initialize() first.")
-
-        # notifications/initialized is a notification (no id)
-        payload = {"jsonrpc": "2.0", "method": "notifications/initialized", "params": {}}
-        # Include session header
-        headers = {"mcp-session-id": self.mcp_session_id}
-        async with self.session.post(self.server_config.url, json=payload, timeout=self.timeout_seconds, headers=headers) as resp:
-            return resp.status, await resp.text()
-
-    async def call(self, method: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
-        if self.session is None:
-            raise RuntimeError("Client session not started. Use 'async with'.")
-        if not self.mcp_session_id:
-            raise RuntimeError("No MCP session established. Call initialize() first.")
-
-        payload = build_json_rpc_request(method, params)
-        headers = {"mcp-session-id": self.mcp_session_id}
-        async with self.session.post(self.server_config.url, json=payload, timeout=self.timeout_seconds, headers=headers) as resp:
-            text = await resp.text()
-            if resp.status != 200:
-                raise RuntimeError(f"HTTP error {resp.status}: {text}")
-            try:
-                data = _parse_streamable_http_body(text, resp.headers.get("Content-Type", ""))
-            except json.JSONDecodeError as exc:
-                raise RuntimeError(f"Failed to parse JSON response: {exc}\nRaw: {text}") from exc
-            return data
-
-
-class MCPStreamClient:
-    """
-    MCP client for connecting to Topcoder MCP server using HTTP.
-    """
-
-    def __init__(self, base_url: Optional[str] = None):
-        self.base_url = base_url or settings.MCP_BASE_URL
-        self._session: Optional[aiohttp.ClientSession] = None
-        self._session_id: Optional[str] = None
-
-    async def _get_session(self) -> aiohttp.ClientSession:
-        if not self._session:
-            headers = {
-                "Content-Type": "application/json",
-                "Accept": "application/json, text/event-stream",
-            }
-            self._session = aiohttp.ClientSession(headers=headers)
-        return self._session
-
-    async def close(self):
-        if self._session:
-            await self._session.close()
-            self._session = None
-
-    async def _rpc(self, method: str, params: Dict[str, Any]) -> Any:
-        session = await self._get_session()
-        payload = {
-            "jsonrpc": "2.0",
-            "id": 1,
-            "method": method,
-            "params": params
-        }
-        
-        async with session.post(self.base_url, json=payload) as resp:
-            # Handle Server-Sent Events (SSE) format
-            if resp.headers.get("content-type", "").startswith("text/event-stream"):
-                async for line in resp.content:
-                    line = line.decode('utf-8').strip()
-                    if line.startswith('data: '):
-                        try:
-                            data = json.loads(line[6:])  # Remove 'data: ' prefix
-                            if isinstance(data, dict) and "result" in data:
-                                return data["result"]
-                            return data
-                        except json.JSONDecodeError:
-                            continue
-                raise Exception("No valid JSON data found in SSE stream")
-            else:
-                # Handle regular JSON response
-                try:
-                    data = await resp.json()
-                except Exception:
-                    text = await resp.text()
-                    raise Exception(f"Failed to parse MCP response ({resp.status}): {text}")
-
-                if resp.status >= 400:
-                    raise Exception(f"MCP error {resp.status}: {data}")
-
-                # Support JSON-RPC style responses or direct result bodies
-                if isinstance(data, dict) and "result" in data:
-                    return data["result"]
-                return data
-
-    async def initialize_session(self) -> str:
-        """Initialize a session with the MCP server."""
-        result = await self._rpc("initialize", {
-            "protocolVersion": "2024-11-05",
-            "capabilities": {
-                "tools": {},
-                "resources": {}
-            },
-            "clientInfo": {
-                "name": "topcoder-mcp-agent",
-                "version": "1.0.0"
-            }
-        })
-        print("DEBUG: Initialize result:", result)
-        return "initialized"
-
-    async def list_resources(self) -> List[Dict[str, Any]]:
-        """List available resources from the MCP server."""
-        result = await self._rpc("resources/list", {})
-        # Normalize result to a list
-        if isinstance(result, dict) and "resources" in result:
-            return result["resources"] or []
-        if isinstance(result, list):
-            return result
-        return []
-
-    async def list_tools(self) -> List[Dict[str, Any]]:
-        """List available tools from the MCP server."""
-        result = await self._rpc("tools/list", {})
-        if isinstance(result, dict) and "tools" in result:
-            return result["tools"] or []
-        if isinstance(result, list):
-            return result
-        return []
-
-
-

backup_v1/prompts/default_responses.py

@@ -1,16 +0,0 @@
-# app/prompts/default_responses.py
-
-default_help_response = (
-    "🤖 I’m a Topcoder MCP Agent! I can:\n"
-    "- Show active challenges\n"
-    "- Fetch member profiles (with handle)\n"
-    "- Answer questions about Topcoder processes\n\n"
-    "I won’t be able to help with general knowledge or personal tasks."
-)
-
-help_triggers = [
-    "what can you do?",
-    "what can you do for me",
-    "help",
-    "how can you help me?",
-]

backup_v1/prompts/system_prompts.py

@@ -1,10 +0,0 @@
-# app/prompts/system_prompts.py
-
-
-topcoder_system_prompt = (
-    "You are a specialized conversational assistant for the Topcoder Member Communication Platform (MCP). "
-    "Your ONLY purpose is to help users with Topcoder-related topics — such as challenges, member profiles, platform usage, rules, or community resources.\n\n"
-    "You MUST NOT answer questions about general knowledge (e.g., history, science, geography), personal advice, current events, or any topic outside Topcoder.\n\n"
-    "If the user asks something unrelated to Topcoder, politely respond that you are only trained to assist with Topcoder-specific information."
-)
-

backup_v1/ui/interface.py

@@ -1,151 +0,0 @@
-import asyncio
-import gradio as gr
-
-from app.agent.model_client import LLMClient
-from app.agent.schema import ToolRequest
-from app.agent.tool_executor import ToolExecutor
-from app.agent.prompt_engine import PromptBuilder
-from app.prompts.system_prompts import topcoder_system_prompt
-from app.prompts.default_responses import default_help_response, help_triggers
-from app.config.static_responses import REJECT_RESPONSE
-from app.utils.json_filter import filter_tool_response
-
-llm_client = LLMClient()
-tool_executor = ToolExecutor()
-prompt_builder = PromptBuilder()
-
-async def agent_response(user_message, history):
-    # STEP 1: Dynamic tool/resource decision using compact specs
-    decision_prompt = prompt_builder.build_tool_decision_prompt(user_message)
-    decision_json = await llm_client.decide_tool(decision_prompt)
-    
-
-    # Handle both tool and resource decisions
-    tool = decision_json.get("tool")
-    resource = decision_json.get("resource")
-    params = decision_json.get("params", {})
-    
-    # For resources, also extract endpoint and method from top level
-    if resource:
-        endpoint = decision_json.get("endpoint")
-        method = decision_json.get("method")
-        if endpoint:
-            params["endpoint"] = endpoint
-        if method:
-            params["method"] = method
-
-    # STEP 2: Clean up parameters (remove empty values)
-    if params:
-        params = {k: v for k, v in params.items() if v and v != ""}
-
-    # STEP 3: General chat
-    if tool == "chat" or resource == "chat":
-        return await llm_client.chat([
-            {"role": "system", "content": topcoder_system_prompt},
-            {"role": "user", "content": user_message}
-        ])
-
-    # STEP 4: Rejection
-    if tool == "reject" or resource == "reject":
-        return REJECT_RESPONSE
-
-    # STEP 5: Determine what to execute
-    available_tools = tool_executor.get_available_tools()
-    available_resources = tool_executor.get_available_resources()
-    
-    if tool and tool in available_tools:
-        # Execute tool
-        target_name = tool
-        target_type = "tool"
-    elif resource and resource in available_resources:
-        # Execute resource
-        target_name = resource
-        target_type = "resource"
-    else:
-        # Invalid selection
-        invalid_name = tool or resource
-        return f"❌ '{invalid_name}' not available. Available tools: {', '.join(available_tools)}, Available resources: {', '.join(available_resources)}"
-
-    # STEP 6: Parameter extraction if needed
-    if not params:
-        if target_type == "tool":
-            param_extraction_prompt = f"""
-Extract parameters for the tool '{target_name}' from this user query: "{user_message}"
-
-Available parameters for {target_name}:
-{tool_executor.get_tool_parameters(target_name)}
-
-Only include parameters that have meaningful values from the user query.
-Respond ONLY with a JSON object containing the extracted parameters:
-{{"params": {{"param1": "value1", "param2": "value2"}} }}
-"""
-            param_response = await llm_client.complete_json(param_extraction_prompt)
-            params = param_response.get("params", {})
-            # Clean up extracted parameters too
-            params = {k: v for k, v in params.items() if v and v != ""}
-        else:  # resource
-            # For resources, we need to extract both endpoint parameters AND API parameters
-            param_extraction_prompt = f"""
-Extract parameters for the resource '{target_name}' from this user query: "{user_message}"
-
-The user wants to call a resource API. You need to extract:
-1. API parameters that should be passed to the endpoint (like handle, challengeId, etc.)
-2. Any path parameters that need to be substituted in the URL
-
-For example, if the user asks for "member abhishekrn", extract:
-- handle: "abhishekrn" (for the API call)
-
-Common parameters to look for:
-- handle: member handle/username
-- challengeId: challenge ID
-- groupId: group ID
-- memberId: member ID
-- Any other IDs or values mentioned in the query
-
-Respond ONLY with a JSON object containing the extracted parameters:
-{{"params": {{"param1": "value1", "param2": "value2"}} }}
-"""
-            param_response = await llm_client.complete_json(param_extraction_prompt)
-            extracted_params = param_response.get("params", {})
-            # Clean up extracted parameters
-            extracted_params = {k: v for k, v in extracted_params.items() if v and v != ""}
-            
-            # Merge with existing params (endpoint and method from LLM decision)
-            params.update(extracted_params)
-
-    # STEP 7: Execute tool or resource using compact specs
-    request = ToolRequest(tool=target_name, params=params)
-    tool_result = await tool_executor.execute(request)
-
-    if tool_result.status != "success":
-        return f"❌ Error executing `{target_name}`: {tool_result.message}"
-
-    # STEP 8: Filter response before summarizing
-    filtered_data = filter_tool_response(target_name, tool_result.data)
-
-    summarization_prompt = f"""
-    You are a Topcoder assistant.
-
-    The user asked: "{user_message}"
-    {target_type.capitalize()} used: {target_name}
-    Response (filtered):
-    {filtered_data}
-
-    Write a clear, concise response that helps the user based on this data.
-    """
-
-    final_response = await llm_client.chat([
-        {"role": "system", "content": "You are a helpful Topcoder assistant summarizing tool/resource outputs."},
-        {"role": "user", "content": summarization_prompt.strip()}
-    ])
-    
-    return final_response
-
-def launch_ui():
-    demo = gr.ChatInterface(
-        fn=agent_response,
-        title="🧠 Topcoder MCP Agent",
-        theme="soft",
-        examples=["Active AI Challenges", "Member details", "Tell me a joke"],
-    )
-    demo.launch()

backup_v1/utils/json_filter.py

@@ -1,72 +0,0 @@
-def filter_tool_response(tool: str, data: dict | list) -> dict | list:
-    """
-    Filters the API response to first 5 items with only relevant fields.
-    """
-    # Handle challenge-related tools and resources
-    if "challenge" in tool.lower() or tool == "get_challenges":
-        if isinstance(data, list):
-            # Return only first 5 challenges with relevant fields
-            return [
-                {
-                    "id": challenge.get("id"),
-                    "name": challenge.get("name"),
-                    "track": challenge.get("track"),
-                    "totalPrizes": challenge.get("totalPrizes", 0),
-                    "status": challenge.get("status"),
-                    "registrationEndDate": challenge.get("registrationEndDate")
-                }
-                for challenge in data[:5]  # First 5 challenges only
-            ]
-        else:
-            # Single challenge - return relevant fields only
-            return {
-                "id": data.get("id"),
-                "name": data.get("name"),
-                "track": data.get("track"),
-                "totalPrizes": data.get("totalPrizes", 0),
-                "status": data.get("status")
-            }
-
-    # Handle member-related tools and resources
-    elif "member" in tool.lower() or tool == "get_member":
-        if isinstance(data, list):
-            # Return only first 5 members with relevant fields
-            return [
-                {
-                    "handle": member.get("handle"),
-                    "firstName": member.get("firstName"),
-                    "lastName": member.get("lastName"),
-                    "status": member.get("status"),
-                    "country": member.get("homeCountryCode")
-                }
-                for member in data[:5]  # First 5 members only
-            ]
-        else:
-            # Single member - return relevant fields only
-            return {
-                "handle": data.get("handle"),
-                "firstName": data.get("firstName"),
-                "lastName": data.get("lastName"),
-                "status": data.get("status"),
-                "country": data.get("homeCountryCode")
-            }
-
-    # Handle skills-related tools
-    elif "skill" in tool.lower():
-        if isinstance(data, list):
-            # Return only first 5 skills with relevant fields
-            return [
-                {
-                    "id": skill.get("id"),
-                    "name": skill.get("name"),
-                    "description": skill.get("description", "")
-                }
-                for skill in data[:5]  # First 5 skills only
-            ]
-        return data
-
-    # Default fallback - limit to first 5 items for any list
-    if isinstance(data, list):
-        return data[:5]
-    
-    return data

config/api_endpoints.py

@@ -9,8 +9,4 @@ API_BASE_URLS = {
 
 def get_base_url(resource_name: str) -> str:
     """Get base URL for a resource."""
-    return API_BASE_URLS.get(resource_name, "")
-
-def get_all_base_urls() -> dict:
-    """Get all available base URLs."""
-    return API_BASE_URLS.copy()
+    return API_BASE_URLS.get(resource_name, "")

config/conversation_config.py

@@ -32,15 +32,12 @@ class EntityExtractionConfig:
     
     # Extraction limits (configurable)
     max_entities_per_tool: int = 5
-    max_member_entities: int = 3
     
     # Entity confidence
     default_entity_confidence: float = 0.8
-    high_confidence: float = 0.9
     
     # Dynamic extraction settings
     use_dynamic_patterns: bool = True
-    extract_from_response_data: bool = True
 
 @dataclass
 class ConversationMemoryConfig:
@@ -57,9 +54,6 @@ class ConversationMemoryConfig:
     # Response truncation
     max_response_storage_chars: int = 1000
     max_context_summary_chars: int = 200
-    
-    # Entity tracking
-    max_tracked_entities: int = 50
 
 @dataclass
 class HistoryHandlerConfig:
@@ -68,7 +62,6 @@ class HistoryHandlerConfig:
     # Response limits
     max_response_chars_for_display: int = 200
     max_turns_in_prompt: int = 5
-    max_entity_contexts: int = 10
     
     # Fallback messages
     no_context_message: str = "I don't see any relevant information from our previous conversation to answer that question. Could you provide more context or ask me to search for new information?"
@@ -81,12 +74,6 @@ class ToolMappingConfig:
     
     # Dynamic mapping settings
     use_runtime_discovery: bool = True
-    cache_mappings: bool = True
-    refresh_cache_on_spec_change: bool = True
-    
-    # Fallback behavior when mapping fails
-    default_entity_type: str = "unknown"
-    log_unmapped_tools: bool = True
 
 # Global configuration instance
 class ConversationConfig:

config/prompt_templates.py

@@ -9,53 +9,6 @@ from typing import Dict, Any
 class PromptTemplates:
     """Centralized prompt templates for the Topcoder MCP Agent."""
     
-    # System prompts
-    TOOL_DECISION_SYSTEM = (
-        "You are a tool decision assistant. Analyze the user's request and available "
-        "tools/resources, then respond ONLY with a JSON object containing the tool "
-        "name and parameters."
-    )
-    
-    JSON_ASSISTANT_SYSTEM = (
-        "You are a strict API assistant. Only return valid JSON."
-    )
-    
-    SUMMARIZATION_SYSTEM = (
-        "You are a helpful Topcoder assistant summarizing tool/resource outputs."
-    )
-    
-    # Enhanced tool parameter extraction templates with context awareness
-    TOOL_PARAM_EXTRACTION = """Extract parameters for the tool '{tool_name}' from this user query: "{user_message}"
-
-Available parameters for {tool_name}:
-{tool_parameters}
-
-CONTEXT-AWARE PARAMETER EXTRACTION:
-- Consider the conversation context when extracting parameters
-- If the user is asking follow-up questions, use context from previous responses
-- If the conversation mentioned specific filters (AI, Machine Learning, etc.), include them
-- If the conversation mentioned specific statuses (Active, Completed, etc.), include them
-- If the user is asking for "more" or "similar" items, use context from previous responses
-- If the user is refining a previous query, combine new parameters with context
-
-EXTRACTION GUIDELINES:
-- Only include parameters that have meaningful values from the user query OR conversation context
-- For follow-up questions, consider what was discussed before
-- If the user mentions "AI challenges" after discussing challenges, include track="AI"
-- If the user asks for "more skills" after discussing programming, include appropriate filters
-- If the user asks for "similar challenges", use the same parameters as before
-
-Respond ONLY with a JSON object containing the extracted parameters:
-{{"params": {{"param1": "value1", "param2": "value2"}} }}"""
-
-    # Enhanced resource parameter extraction templates with context awareness
-    RESOURCE_PARAM_EXTRACTION = """Extract parameters for '{resource_name}' from: "{user_message}"
-
-Extract only parameters with meaningful values from the user query.
-Follow the parameter descriptions exactly as specified in the API specs.
-
-Respond ONLY with: {{"params": {{"param1": "value1"}} }}"""
-
     # Gradio interface specific prompts with context awareness
     GRADIO_TOOL_PARAM_EXTRACTION = """Extract parameters for the tool '{tool_name}' from this user query: "{user_message}"
 
@@ -112,110 +65,6 @@ INSTRUCTIONS:
 - Keep it simple and helpful - don't expose technical details
 - Remember: "user" is a search term, not a field name"""
 
-    # Tool decision prompt template with conversation context support
-    TOOL_DECISION_PROMPT = """You are an assistant that ONLY works with Topcoder's Member Communication Platform (MCP).
-
-Available MCP Tools:
-{tools_list}
-
-Available MCP Resources (APIs):
-{resources_list}
-
-User query: "{user_message}"
-
-{conversation_context}
-
-Analyze the query and determine:
-1. If it's asking for Topcoder data (challenges, members, skills, etc.) → select appropriate tool/resource
-2. If it's a brief greeting, thanks, or simple conversational response → use "chat"
-3. If it's asking for general knowledge, non-Topcoder information, or complex non-Topcoder topics → use "reject"
-
-CONTEXT-AWARE DECISION MAKING:
-- Consider the conversation context when making decisions
-- If the user is asking follow-up questions, understand they may be referring to previous entities
-- If the conversation has established context (e.g., discussing AI challenges), use that context
-- For pronouns like "that", "it", "those", refer to the conversation context
-- For requests like "more", "similar", "other", use context from previous interactions
-
-IMPORTANT: 
-- For tools: Only include parameters that have meaningful values from the user query or conversation context
-- For resources: Use the resource name and specify the endpoint path and method
-- For member queries: ALWAYS use "Member_V6_API_Swagger" as a RESOURCE (not tool) with endpoint "/members" 
-- For challenge queries: Use "query-tc-challenges" as a TOOL with parameters like status, track, page, perPage
-- For skills queries: Use "query-tc-skills" as a TOOL with parameters like name, skillId, page, perPage
-- Always include relevant parameters in the params object for API calls
-
-Respond ONLY with a JSON object in this format:
-{{"tool": "<tool_name>", "params": {{"param1": "value1", "param2": "value2"}} }}
-OR
-{{"resource": "<resource_name>", "endpoint": "/path", "method": "GET", "params": {{"param1": "value1", "param2": "value2"}} }}
-
-Choose either "tool" OR "resource", not both.
-
-"""
-
-    # Response summarization template
-    RESPONSE_SUMMARIZATION = """You are a Topcoder assistant.
-
-The user asked: "{user_message}"
-{target_type} used: {target_name}
-Response (filtered):
-{filtered_data}
-
-Write a clear, concise response that helps the user based on this data."""
-
-    @classmethod
-    def get_tool_param_extraction_prompt(cls, tool_name: str, user_message: str, tool_parameters: str, conversation_context: str = None) -> str:
-        """Get formatted tool parameter extraction prompt with optional conversation context."""
-        base_prompt = cls.TOOL_PARAM_EXTRACTION.format(
-            tool_name=tool_name,
-            user_message=user_message,
-            tool_parameters=tool_parameters
-        )
-        
-        if conversation_context:
-            base_prompt += f"\n\nCONVERSATION CONTEXT:\n{conversation_context}\n\nUse this context to better understand the user's intent and extract appropriate parameters."
-        
-        return base_prompt
-    
-    @classmethod
-    def get_resource_param_extraction_prompt(cls, resource_name: str, user_message: str, conversation_context: str = None) -> str:
-        """Get formatted resource parameter extraction prompt with optional conversation context."""
-        base_prompt = cls.RESOURCE_PARAM_EXTRACTION.format(
-            resource_name=resource_name,
-            user_message=user_message
-        )
-        
-        if conversation_context:
-            base_prompt += f"\n\nCONVERSATION CONTEXT:\n{conversation_context}\n\nUse this context to better understand the user's intent and extract appropriate parameters."
-        
-        return base_prompt
-    
-    @classmethod
-    def get_tool_decision_prompt(cls, user_message: str, tools_list: str, resources_list: str, conversation_context: str = None) -> str:
-        """Get formatted tool decision prompt with optional conversation context."""
-        context_section = ""
-        if conversation_context:
-            context_section = f"CONVERSATION CONTEXT:\n{conversation_context}\n"
-        
-        return cls.TOOL_DECISION_PROMPT.format(
-            user_message=user_message,
-            tools_list=tools_list,
-            resources_list=resources_list,
-            conversation_context=context_section
-        )
-    
-    @classmethod
-    def get_response_summarization_prompt(cls, user_message: str, target_type: str, 
-                                        target_name: str, filtered_data: Any) -> str:
-        """Get formatted response summarization prompt."""
-        return cls.RESPONSE_SUMMARIZATION.format(
-            user_message=user_message,
-            target_type=target_type,
-            target_name=target_name,
-            filtered_data=filtered_data
-        )
-
     # Gradio interface specific methods
     @classmethod
     def get_gradio_tool_param_extraction_prompt(cls, tool_name: str, user_message: str, tool_parameters: str, conversation_context: str = None) -> str:

config/static_responses.py

@@ -4,32 +4,3 @@ REJECT_RESPONSE = (
     "I focus on Topcoder-related topics like challenges, members, and skills. "
     "What would you like to know about Topcoder?"
 )
-
-# Human-readable error messages for users
-INVALID_TOOL_RESPONSE = (
-    "I'm sorry, but I couldn't find the specific tool you're looking for. "
-    "I can help you with Topcoder challenges, member information, skills, and other platform features. "
-    "Could you please rephrase your request or ask about something else related to Topcoder?"
-)
-
-INVALID_RESOURCE_RESPONSE = (
-    "I'm sorry, but I couldn't access the specific resource you requested. "
-    "I can help you with Topcoder challenges, member profiles, skills, and other platform information. "
-    "Could you please try asking about something else related to Topcoder?"
-)
-
-GENERAL_ERROR_RESPONSE = (
-    "I encountered an issue while processing your request. "
-    "Please try again or ask about something else related to Topcoder."
-)
-
-TOOL_EXECUTION_ERROR_RESPONSE = (
-    "I'm sorry, but I encountered an error while trying to get that information for you. "
-    "Please try again or ask about something else related to Topcoder."
-)
-
-# Console logging messages for developers
-CONSOLE_INVALID_TOOL_MSG = "❌ Invalid tool selection: '{tool_name}' not available. Available tools: {available_tools}"
-CONSOLE_INVALID_RESOURCE_MSG = "❌ Invalid resource selection: '{resource_name}' not available. Available resources: {available_resources}"
-CONSOLE_TOOL_EXECUTION_ERROR_MSG = "❌ Error executing tool '{tool_name}': {error_message}"
-CONSOLE_RESOURCE_EXECUTION_ERROR_MSG = "❌ Error executing resource '{resource_name}': {error_message}"

src/application/services/prompt_service.py

@@ -1,5 +1,4 @@
 from src.mcp.compact_utils import CompactSpecsUtils
-from config.prompt_templates import PromptTemplates
 from typing import List, Dict, Any, Optional
 
 class PromptBuilder:

src/infrastructure/providers/llm_provider.py

@@ -3,7 +3,6 @@ import json
 from typing import List, Dict, Any, Optional
 from config import settings
 from config.system_prompts import topcoder_system_prompt
-from config.prompt_templates import PromptTemplates
 
 class LLMClient:
     def __init__(self):