from __future__ import annotations import logging from typing import Any, Protocol from mcp.server import Server from mcp.types import ( CallToolResult, GetPromptResult, ListPromptsResult, ListResourcesResult, ListToolsResult, ReadResourceResult, TextContent, TextResourceContents, Resource, Tool, Prompt ) from .config import Config from .prompts import PromptsManager from . import __version__ NAME = "mini-db-mcp" logger = logging.getLogger(__name__) class ResourcesProtocol(Protocol): async def list_resources(self) -> list[Resource]: ... async def read_resource(self, uri: str) -> str: ... class ToolsProtocol(Protocol): async def list_tools(self) -> list[Tool]: ... async def call_tool(self, name: str, arguments: dict[str, Any]) -> str: ... class PromptsProtocol(Protocol): async def list_prompts(self) -> list[Prompt]: ... async def get_prompt( self, name: str, arguments: dict[str, Any] ) -> GetPromptResult: ... class MiniServer(): def __init__(self, config: Config) -> None: self.config = config self.prompts_manager = PromptsManager() self.server = _create_mcp_server( prompts_manager=self.prompts_manager ) async def start_stdio(self) -> None: from mcp.server.stdio import stdio_server async with stdio_server() as streams: read_stream, write_stream = streams await self.server.run( read_stream, write_stream, self.server.create_initialization_options(), ) def _create_mcp_server( *, prompts_manager: PromptsProtocol | None = None, ) -> Server: async def on_list_prompts(ctx, params): return ListPromptsResult(prompts=await prompts_manager.list_prompts()) async def on_get_prompt(ctx, params): return await prompts_manager.get_prompt(params.name, dict(params.arguments or {})) server = Server( name=NAME, version=__version__, on_list_prompts=on_list_prompts, on_get_prompt=on_get_prompt, ) return server