Encoding DevOps MCP Server

import webbrowser from urllib.parse import quote from mcp.server.fastmcp import Context from encoding_devops.mcp_instance import AppContext, mcp @mcp.tool() async def get_job_by_name(name: str, ctx: Context) -> str: """Get details of an encoding job by its name""" app_ctx: AppContext = ctx.request_context.lifespan_context job_data = await app_ctx.client.get_job_by_name(name) return job_data @mcp.tool() async def get_job_tasks_by_id(job_id: str, ctx: Context) -> str: """Get tasks for a specific job by its ID""" app_ctx: AppContext = ctx.request_context.lifespan_context tasks_data = await app_ctx.client.get_job_tasks_by_id(job_id) return tasks_data @mcp.tool() async def get_clients(ctx: Context) -> str: """Get list of all clients""" app_ctx: AppContext = ctx.request_context.lifespan_context clients_data = await app_ctx.client.get_clients() return clients_data @mcp.tool() async def is_cluster_busy(ctx: Context) -> str: """Check if the encoding cluster is busy (has jobs in progress)""" app_ctx: AppContext = ctx.request_context.lifespan_context jobs_count = await app_ctx.client.get_inprogress_jobs_count() is_busy = jobs_count > 0 return {"is_busy": is_busy, "jobs_count": jobs_count, "status": "busy" if is_busy else "not busy"} @mcp.tool() async def get_latest_jobs(limit: int, ctx: Context) -> str: """ Get the most recent encoding jobs Args: limit: Number of jobs to return (default: 3, max: 10) Returns: String representation of the latest jobs """ if not 1 <= limit <= 10: return "Error: Limit must be between 1 and 10 jobs" app_ctx: AppContext = ctx.request_context.lifespan_context jobs_data = await app_ctx.client.get_latest_jobs(limit) return {"count": len(jobs_data) if jobs_data else 0, "limit": limit, "jobs": jobs_data if jobs_data else []} @mcp.tool() async def search_movie(title: str, ctx: Context) -> str: """ Search for a movie by title Args: title: Movie title to search for Returns: String representation of the search results """ app_ctx: AppContext = ctx.request_context.lifespan_context results = await app_ctx.omdb_client.search_movie(title) return {"query": title, "total_results": int(results.get("totalResults", 0)), "movies": results.get("Search", [])} @mcp.tool() async def get_movie_details(imdb_id: str, ctx: Context) -> str: """ Get detailed information about a movie by its IMDB ID Args: imdb_id: IMDB ID of the movie (e.g., 'tt0111161') Returns: String representation of the movie details """ app_ctx: AppContext = ctx.request_context.lifespan_context movie_data = await app_ctx.omdb_client.get_movie_details(imdb_id) if not movie_data: return {"error": f"No movie found with IMDB ID: {imdb_id}"} return movie_data @mcp.tool() async def open_email( body: str, to: str = "", subject: str = "", cc: str = "", bcc: str = "", ctx: Context = None ) -> str: """ Opens the default email client with your message ready to go Args: body: What you want to say in the email to: Who to send it to (optional) subject: Email subject line (optional) cc: CC recipients (optional) bcc: BCC recipients (optional) ctx: MCP context Returns: str: Quick status message so you know it worked """ # URL encode everything (handles special characters and spaces) params = [] # Start with the body since it's required params.append(f"body={quote(body)}") # Add any optional fields that were provided if subject: params.append(f"subject={quote(subject)}") if cc: params.append(f"cc={quote(cc)}") if bcc: params.append(f"bcc={quote(bcc)}") # Build the mailto URL - if we have a recipient, include it in the base URL mailto_base = f"mailto:{quote(to)}?" if to else "mailto:?" mailto_url = mailto_base + "&".join(params) # Fire up the email client webbrowser.open(mailto_url, new=1) return "Ready to go! Your email client should be open with the template loaded."