""" MCP Client Service for Yuga Planner This service handles Model Context Protocol client operations for scheduling tools. Provides a clean interface for integrating MCP scheduling functionality. """ import base64 from typing import Dict, Any from utils.logging_config import setup_logging, get_logger from handlers.mcp_backend import process_message_and_attached_file # Initialize logging setup_logging() logger = get_logger(__name__) class MCPClientService: """Service for MCP client operations and scheduling tool integration""" def __init__(self): self.tools = [ { "type": "function", "function": { "name": "schedule_tasks_with_calendar", "description": "Create an optimized schedule by analyzing calendar events and breaking down tasks. Upload a calendar .ics file and describe the task you want to schedule.", "parameters": { "type": "object", "properties": { "task_description": { "type": "string", "description": "Description of the task or project to schedule (e.g., 'Create a new EC2 instance on AWS')", }, "calendar_file_content": { "type": "string", "description": "Base64 encoded content of the .ics calendar file, or 'none' if no calendar provided", }, }, "required": ["task_description", "calendar_file_content"], }, }, } ] async def call_scheduling_tool( self, task_description: str, calendar_file_content: str ) -> Dict[str, Any]: """ Call the scheduling backend tool. Args: task_description: Description of the task to schedule calendar_file_content: Base64 encoded calendar content or 'none' Returns: Dict containing the scheduling result """ try: if calendar_file_content.lower() == "none": file_content = b"" else: file_content = base64.b64decode(calendar_file_content) logger.debug(f"Calling MCP backend with task: {task_description}") result = await process_message_and_attached_file( file_content=file_content, message_body=task_description, file_name="calendar.ics", ) logger.debug( f"MCP backend returned status: {result.get('status', 'unknown')}" ) return result except Exception as e: logger.error(f"Error calling scheduling tool: {e}") return {"error": str(e), "status": "failed"}