init

src/mini_db_mcp/server.py

@@ -0,0 +1,88 @@
+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
+