MCP SQL Server

"""Base class for MCP tools/services.""" from abc import ABC, abstractmethod from typing import Any, Optional try: from fastmcp import Context except ImportError: from ..dependencies import ensure_deps_once ensure_deps_once() from fastmcp import Context from ..credentials import CredentialsManager from ..connection import ConnectionManager from ..inspector import DatabaseInspector from ..executor import QueryExecutor class MCPTool(ABC): """Abstract base class for MCP tools.""" def __init__( self, connection_manager: ConnectionManager, credentials_manager: CredentialsManager, inspector: DatabaseInspector, executor: QueryExecutor ): self.conn_manager = connection_manager self.creds_manager = credentials_manager self.inspector = inspector self.executor = executor @property @abstractmethod def name(self) -> str: """Tool name (used for registration).""" pass @property @abstractmethod def description(self) -> str: """Tool description (used in MCP metadata).""" pass @abstractmethod async def execute(self, ctx: Context, *args, **kwargs) -> Any: """Execute the tool logic. Note: Implementations should define explicit parameters instead of **kwargs. FastMCP requires explicit parameter definitions for tool registration. """ pass def validate_credentials(self, ctx: Context, **kwargs) -> tuple[bool, Optional[str]]: """Validate that required credentials are present. Returns: tuple: (is_valid, error_message) """ creds = self.creds_manager.get_from_context( ctx, user=kwargs.get('user'), password=kwargs.get('password'), server=kwargs.get('server_name'), database=kwargs.get('database'), driver=kwargs.get('driver'), port=kwargs.get('port') ) if not creds.is_valid(): return False, "Missing credentials (user, password, or server)" return True, None