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())