MCP Server Modal

  • src
  • modal_server
import json import logging from collections.abc import Sequence import subprocess from typing import Any import httpx import asyncio from dotenv import load_dotenv from mcp.server import Server from mcp.types import ( Resource, Tool, TextContent, ImageContent, EmbeddedResource, ) import mcp.types as types from mcp.server import Server from pydantic import AnyUrl import mcp.server.stdio from pydantic import AnyUrl # Load environment variables load_dotenv() # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger("modal-server") notes: dict[str, str] = {} app = Server("modal-server") @app.list_resources() async def list_resources() -> list[Resource]: """List available resources.""" return [] @app.read_resource() async def handle_read_resource(uri: AnyUrl) -> str: try: return json.dumps({"result": "example"}, indent=2) except httpx.HTTPError as e: raise RuntimeError(f"API error: {str(e)}") @app.list_prompts() async def handle_list_prompts() -> list[types.Prompt]: """ List available prompts. Each prompt can have optional arguments to customize its behavior. """ return [ types.Prompt( name="summarize-notes", description="Creates a summary of all notes", arguments=[ types.PromptArgument( name="style", description="Style of the summary (brief/detailed)", required=False, ) ], ) ] @app.get_prompt() async def handle_get_prompt( name: str, arguments: dict[str, str] | None ) -> types.GetPromptResult: """ Generate a prompt by combining arguments with server state. The prompt includes all current notes and can be customized via arguments. """ if name != "summarize-notes": raise ValueError(f"Unknown prompt: {name}") style = (arguments or {}).get("style", "brief") detail_prompt = " Give extensive details." if style == "detailed" else "" return types.GetPromptResult( description="Summarize the current notes", messages=[ types.PromptMessage( role="user", content=types.TextContent( type="text", text=f"Here are the current notes to summarize:{detail_prompt}\n\n" + "\n".join( f"- {name}: {content}" for name, content in notes.items() ), ), ) ], ) @app.list_tools() async def handle_list_tools() -> list[types.Tool]: """ List available tools. Each tool specifies its arguments using JSON Schema validation. """ return [ Tool( name="deploy", description="some description", inputSchema={ "type": "object", "properties": { "modal_path": {"type": "string"}, }, "required": ["message"], }, ) ] @app.call_tool() async def call_tool( name: str, arguments: Any ) -> Sequence[TextContent | ImageContent | EmbeddedResource]: """Handle tool calls for weather forecasts.""" if name != "deploy": raise ValueError(f"Unknown tool: {name}") if not isinstance(arguments, dict) or "modal_path" not in arguments: raise ValueError("Invalid forecast arguments") modal_path = arguments["modal_path"] try: res = deploy(modal_path) return [ TextContent(type="text", text=json.dumps(f"Deploy result: {res}", indent=2)) ] except httpx.HTTPError as e: raise RuntimeError(f"Ran in error: {str(e)}") def deploy(modal_path: str = "model_app.py") -> str: """ Deploy a model using Modal CLI command. Args: modal_path: Path to the modal file to deploy Returns: str: deployment result """ try: # Run modal deploy command process = subprocess.run(["modal", "deploy", modal_path], capture_output=True, text=True) # Check if the command was successful if process.returncode == 0: return f"Deploy success: {process.stdout}" else: raise RuntimeError(f"Deploy failed: {process.stderr}") # if process.returncode == 0: # message = f"Deployment successful: {stdout.decode()}" # else: # message = f"Deployment failed: {stderr.decode()}" # return message except Exception as e: return f"Deployment error: {str(e)}" async def main(): from mcp.server.stdio import stdio_server async with stdio_server() as (read_stream, write_stream): await app.run(read_stream, write_stream, app.create_initialization_options())