SQL Server MCP

Instructions

Implementation Reference

• src/mssql_mcp/server.py:119-148 (handler)

  The MCP tool handler function for 'connect' that establishes the database connection via ConnectionManager and returns status details.
  @mcp.tool() def connect() -> dict[str, Any]: """Establish connection to the SQL Server database. Uses configuration from environment variables: - MSSQL_HOST: Server hostname or IP - MSSQL_USER: Username - MSSQL_PASSWORD: Password - MSSQL_DATABASE: Database name - MSSQL_PORT: Port (default: 1433) Returns: Connection status and details including host, database, and timestamp. """ try: manager = get_connection_manager() info = manager.connect() return { "status": "connected", "host": info.host, "database": info.database, "connected_at": info.connected_at.isoformat(), } except ConnectionError as e: return {"status": "error", "error": str(e)} except Exception as e: logger.error(f"Unexpected error during connect: {e}") return {"status": "error", "error": f"Unexpected error: {e}"}
• src/mssql_mcp/connection.py:59-103 (helper)

  Core implementation of database connection establishment in ConnectionManager.connect(), called by the tool handler.
  def connect(self, name: str = "default") -> ConnectionInfo: """Establish a connection to the SQL Server. Args: name: Connection name for reference Returns: ConnectionInfo with connection details Raises: ConnectionError: If connection fails """ if name in self._connections and self._connections[name].is_active: logger.info(f"Reusing existing connection '{name}'") return self._connections[name] try: logger.info(f"Connecting to {self._config.host}/{self._config.database}") self._connection = pymssql.connect( server=self._config.host, user=self._config.user, password=self._config.password, database=self._config.database, port=self._config.port, login_timeout=self._config.timeout, timeout=self._config.query_timeout, ) info = ConnectionInfo( name=name, host=self._config.host, database=self._config.database, connected_at=datetime.now(), is_active=True, ) self._connections[name] = info logger.info(f"Connected successfully to {self._config.database}") return info except pymssql.Error as e: logger.error(f"Connection failed: {e}") raise ConnectionError(f"Failed to connect to {self._config.host}: {e}") from e
• src/mssql_mcp/server.py:31-42 (helper)

  Utility function to retrieve or initialize the global ConnectionManager instance used by the connect tool.
  def get_connection_manager() -> ConnectionManager: """Get or create the global connection manager. Returns: ConnectionManager instance configured from environment variables """ global _connection_manager if _connection_manager is None: config = MSSQLConfig() _connection_manager = ConnectionManager(config) return _connection_manager